qpanda3_runtime.access_service.remote_request#

Remote request module for quantum computing services.

This package provides classes and utilities for constructing and sending requests to remote quantum computing services, including task message builders and asynchronous request workers.

Modules#

chip_task_msg: Message builders for quantum computing tasks.
remote_request_worker: Asynchronous worker for handling quantum service requests.

Submodules#

Classes#

`ChipSampleTaskMsg`	Message builder for quantum chip sampling tasks.
`ChipEstimateTaskMsg`	Message builder for quantum chip estimation tasks.
`RequestWorker`	Threaded worker for handling asynchronous quantum service requests.

Package Contents#

class qpanda3_runtime.access_service.remote_request.ChipSampleTaskMsg#

Message builder for quantum chip sampling tasks.

This class constructs messages for submitting quantum circuit sampling tasks to remote quantum computing services.

Variables:: _msg (dict) – Internal message dictionary containing task configuration.

build(progs, chip_id, shot, is_amend, is_mapping, is_optimization,: specified_block, priority, point_label, token, task_describe)

Build a complete sampling task message.

msg_dict()#: Get the message as a dictionary.

msg_json()#: Get the message as a JSON string.

from_json_str(json_str)#: Load message from JSON string.

is_equal_with(other)#: Compare equality with another message instance.

msg()#

Get the internal message dictionary.

Returns:: Current message dictionary.
Return type:: dict

build(progs: List[str], chip_id: str, shot: int, is_amend: bool, is_mapping: bool, is_optimization: bool, specified_block: [int], priority: int, point_label: int, token: str, task_describe: str)#

Build a sampling task message with all required parameters.

Parameters:

progs (List[str]) – List of quantum programs/circuits to execute.
chip_id (str) – Identifier of the target quantum chip.
shot (int) – Number of measurement shots to perform.
is_amend (bool) – Whether to apply amplitude correction.
is_mapping (bool) – Whether to apply qubit mapping.
is_optimization (bool) – Whether to apply circuit optimization.
specified_block (List[int]) – List of specific blocks to execute on.
priority (int) – Task priority level.
point_label (int) – Label for calibration point.
token (str) – Authentication token.
task_describe (str) – Human-readable task description.

Returns:

Complete task message dictionary.

Return type:

dict

Notes

The message structure follows the PilotOS API specification for quantum sampling tasks.

msg_dict()#

Get the message as a dictionary.

Returns:: Message dictionary.
Return type:: dict

msg_json()#

Get the message as a JSON string.

Returns:: JSON-encoded message string.
Return type:: str

from_json_str(json_str: str)#

Load message from JSON string.

Parameters:: json_str (str) – JSON string containing the task message.

Notes

This method parses the JSON and populates the internal message dictionary with the extracted data.

is_equal_with(other)#

Compare equality with another message instance.

Parameters:: other (ChipSampleTaskMsg) – Another message instance to compare with.
Returns:: True if messages are equal, False otherwise.
Return type:: bool

Notes

This method performs deep comparison of all message fields, including nested configuration data.

class qpanda3_runtime.access_service.remote_request.ChipEstimateTaskMsg#

Message builder for quantum chip estimation tasks.

This class constructs messages for submitting quantum expectation value estimation tasks to remote quantum computing services.

Variables:: _msg (dict) – Internal message dictionary containing task configuration.

build(progs, hamiltonian, qubits, shot, chip_id, is_amend, is_mapping,: is_optimization, specified_block, task_describe, token)

Build a complete estimation task message.

from_json_str(json_str)#: Load message from JSON string.

is_equal_with(other)#: Compare equality with another message instance.

update_hamiltonian(ham)#: Update the Hamiltonian in the message.

update_progs(progs)#: Update the quantum programs in the message.

msg_dict()#: Get the message as a dictionary.

msg_json()#: Get the message as a JSON string.

build(progs: [str], hamiltonian, qubits: List[int], shot: int, chip_id: str, is_amend: bool, is_mapping: bool, is_optimization: bool, specified_block: List[int], task_describe: str, token: str)#

Build an estimation task message with all required parameters.

Parameters:

progs (List[str]) – List of quantum programs/circuits to execute.
hamiltonian – Hamiltonian operator for expectation value calculation.
qubits (List[int]) – List of qubit indices involved in the calculation.
shot (int) – Number of measurement shots to perform.
chip_id (str) – Identifier of the target quantum chip.
is_amend (bool) – Whether to apply amplitude correction.
is_mapping (bool) – Whether to apply qubit mapping.
is_optimization (bool) – Whether to apply circuit optimization.
specified_block (List[int]) – List of specific blocks to execute on.
task_describe (str) – Human-readable task description.
token (str) – Authentication token.

Returns:

Complete task message dictionary.

Return type:

dict

Notes

The message structure follows the PilotOS API specification for quantum expectation value estimation tasks.

from_json_str(json_str: str)#

Load message from JSON string.

Parameters:: json_str (str) – JSON string containing the task message.

Notes

This method parses the JSON and populates the internal message dictionary with the extracted data.

is_equal_with(other)#

Compare equality with another message instance.

Parameters:: other (ChipEstimateTaskMsg) – Another message instance to compare with.
Returns:: True if messages are equal, False otherwise.
Return type:: bool

Notes

This method performs deep comparison of all message fields, including nested configuration data.

update_hamiltonian(ham)#

Update the Hamiltonian in the message.

Parameters:: ham – New Hamiltonian operator to set.

update_progs(progs)#

Update the quantum programs in the message.

Parameters:: progs (List[str]) – New list of quantum programs to set.

msg_dict()#

Get the message as a dictionary.

Returns:: Message dictionary.
Return type:: dict

msg_json()#

Get the message as a JSON string.

Returns:: JSON-encoded message string.
Return type:: str

class qpanda3_runtime.access_service.remote_request.RequestWorker(base_url, outbound_raw_requests_logfile)#

Bases: threading.Thread

Threaded worker for handling asynchronous quantum service requests.

This class runs an asyncio event loop in a separate thread to handle concurrent HTTP requests to quantum computing services. It provides synchronous interfaces that internally use asynchronous operations.

Variables:

loop (asyncio.AbstractEventLoop) – Asyncio event loop running in the worker thread.
_http_client (HTTPClient) – HTTP client for making requests to quantum services.
_base_url (str) – Base URL for the quantum service API.
_outbound_raw_requests_logfile (str) – Path to log file for recording outbound requests.

login(msg, channel)#: Authenticate with the quantum service.

query_task(channel, task_id, token, additional_info)#: Query status of submitted tasks.

list_devices(channel, target)#: List available quantum devices.

query_device_config(channel, msg)#: Query configuration of a quantum device.

estimate(channel, msg)#: Submit an expectation value estimation task.

sample(channel, msg)#: Submit a circuit sampling task.

circuit_chunking_submit(channel, circuits, common_msg, task_type,: max_byte_num_per_packet, max_cir_num_per_packet, encode_type)

Submit large circuits in chunks.

sample_program_set(channel, msg)#: Submit a program set for sampling.

apply_vqsession(channel, msg)#: Apply for a variational quantum session.

release_session(channel, msg)#: Release a variational quantum session.

run_vqtask(channel, msg)#: Run a variational quantum task.

close()#: Stop the worker thread.

loop = None#

login(msg: dict, channel: str = default_channel)#

Authenticate with the quantum service.

Parameters:

msg (dict) – Authentication message containing credentials.
channel (str, optional) – Communication channel identifier (default: default_channel).

Returns:

Authentication response from the server.

Return type:

dict

Notes

This method sends login credentials to obtain an authentication token for subsequent API calls.

query_task(channel: str, task_id: list, token: str, additional_info: List[str] | None = None, timeout: float = 600)#

Query status of submitted tasks.

Parameters:

channel (str) – Communication channel identifier.
task_id (list) – List of task IDs to query.
token (str) – Authentication token.
additional_info (Optional[List[str]], optional) – Additional information fields to request (default: None).

Returns:

List of task status responses.

Return type:

list

Notes

This method queries the status of multiple tasks concurrently and returns their results.

list_devices(token: str, channel=default_channel, target: int | str = '')#

List available quantum devices.

Parameters:

channel (str, optional) – Communication channel identifier (default: default_channel).
target (Union[int, str], optional) – Target device identifier or type (default: ‘’).

Returns:

List of available quantum devices and their properties.

Return type:

dict

Notes

This method retrieves information about quantum computing devices available through the service.

query_device_config(channel: str, msg: dict)#

Query configuration of a quantum device.

Parameters:

channel (str) – Communication channel identifier.
msg (dict) – Message containing device query parameters.

Returns:

Device configuration information.

Return type:

dict

estimate(channel, msg)#

Submit an expectation value estimation task.

Parameters:

channel (str) – Communication channel identifier.
msg (dict) – Task message containing circuit and Hamiltonian information.

Returns:

Task submission response.

Return type:

dict

sample(channel, msg)#

Submit a circuit sampling task.

Parameters:

channel (str) – Communication channel identifier.
msg (dict) – Task message containing circuit information.

Returns:

Task submission response.

Return type:

dict

circuit_chunking_submit(channel, circuits: List[str], common_msg, task_type, max_byte_num_per_packet, max_cir_num_per_packet, encode_type='ascii')#

Submit large circuits in chunks.

Parameters:

channel (str) – Communication channel identifier.
circuits (List[str]) – List of quantum circuits to submit.
common_msg (dict) – Common message parameters for all circuits.
task_type (str) – Type of task (‘sample’ or other supported types).
max_byte_num_per_packet (int) – Maximum bytes per network packet.
max_cir_num_per_packet (int) – Maximum circuits per network packet.
encode_type (str, optional) – Encoding type for circuit strings (default: ‘ascii’).

Returns:

List of submission results for each chunk.

Return type:

list

Notes

This method splits large circuit submissions into smaller chunks to avoid network packet size limitations.

sample_program_set(channel, msg)#

apply_vqsession(channel, msg)#

release_session(channel, msg)#

run_vqtask(channel, msg)#

async circuit_chunking_submit_(channel, circuits: List[str], common_msg, task_type, max_byte_num_per_packet, max_cir_num_per_packet, encode_type='ascii')#

async query_task_(channel: str, task_id: int, token: str, additional_info: List[str] | None = None)#

async query_task_async(channel: str, task_id: list, token: str, additional_info: List[str] | None = None, timeout: float = 600)#

static handle_task_result(async_future, threading_future)#

run()#: Main thread entry point for the worker.

Notes

This method sets up the asyncio event loop and HTTP client, then runs the event loop indefinitely to process asynchronous tasks.

execute(data)#

close()#: Stop the worker thread.

Notes

This method safely stops the asyncio event loop, allowing the worker thread to terminate gracefully.