wy_qcos.engine package

Subpackages

Submodules

wy_qcos.engine.common module

wy_qcos.engine.common.convert_log_format_to_loguru(log_format)

Convert standard Python logging format to loguru format.

参数:

log_format -- Standard Python logging format string

返回:

Loguru format string e.g., "{time:YYYY-MM-DD HH:mm:ss} | {level} | {name}:{line} {message}"

wy_qcos.engine.common.init_logger(log_file_path, debug=False, log_format=None, log_rotate_max_size_mb=None, log_rotate_backup_count=None, log_rotate_compression=None)

Initialize logger with configurable parameters.

参数:

  • log_file_path -- Path to the log file

  • debug -- Enable debug logging level

  • log_format -- Custom log format (defaults to Config.LOG.LOG_FORMAT) Standard Python logging format string

  • log_rotate_max_size_mb -- Max size for log rotation in MB

  • log_rotate_backup_count -- Number of backup log files to retain

  • log_rotate_compression -- Enable gzip compression for rotated logs

wy_qcos.engine.device_mgr_engine module

wy_qcos.engine.device_mgr_engine.call_device_method(driver, method, *args, **kwargs)

Dynamic call method in driver.

参数:

  • driver -- driver

  • method -- the method called

  • args -- args

  • kwargs -- kwargs

返回:

results

wy_qcos.engine.device_mgr_engine.device_manager_flow(device_mgr_info)

Device manager flow.

参数:

device_mgr_info -- device manager info

返回:

None

wy_qcos.engine.device_monitor_engine module

wy_qcos.engine.device_monitor_engine.device_monitor_flow(device_monitor_info)

Device monitor flow.

参数:

device_monitor_info -- device info

返回:

None

wy_qcos.engine.job_engine module

class wy_qcos.engine.job_engine.AggregationInput(*args, **kwargs)

基类：RunInput

参数:

  • args (Any)

  • kwargs (Any)

返回类型:

Any

is_parent: bool
sub_jobs: dict | None = None
sub_results: list[Any] | None = None
class wy_qcos.engine.job_engine.SourceCodeInfo

基类：object

aggregation_type: str
src_code_list: list[dict]
wy_qcos.engine.job_engine.init_driver(driver_class_info, driver_options=None, device=None, job_info=None)

Init driver from driver_class_info.

参数:

  • driver_class_info -- driver class info

  • driver_options -- driver options

  • device -- device info

  • job_info -- job info

返回:

driver

wy_qcos.engine.job_engine.init_transpiler(transpiler_class_info, transpiler_options)

Init transpiler instance.

参数:

  • transpiler_class_info -- transpiler class info

  • transpiler_options -- transpiler options

返回:

transpiler

wy_qcos.engine.job_engine.task_monitor(monitor_info)
wy_qcos.engine.job_engine.parse(src_code_dict, transpiler, code_type)

Parse task.

参数:

  • src_code_dict -- src_code_dict

  • transpiler -- transpiler

  • code_type (str) -- code type

返回:

parsed results

wy_qcos.engine.job_engine.transpile(parsed_gates, driver, transpiler)

Transpile task.

参数:

  • parsed_gates -- parsed gates

  • driver -- driver

  • transpiler -- transpiler

返回:

basis gate list

wy_qcos.engine.job_engine.driver_run(job_info, driver, num_qubits, data)

Driver: run job.

参数:

  • job_info -- job info

  • driver -- driver

  • num_qubits -- number of qubits

  • data -- data

返回:

results

wy_qcos.engine.job_engine.driver_cancel(job_id, driver)

Driver: cancel job.

参数:

  • job_id -- job id

  • driver -- driver

wy_qcos.engine.job_engine.register_signals(job_id, monitor)

Register signal handlers.

参数:

  • job_id -- job id

  • monitor -- monitor

wy_qcos.engine.job_engine.update_progress(artifact_id, progress)

Update progress.

参数:

  • artifact_id -- artifact id

  • progress -- progress

wy_qcos.engine.job_engine.create_src_code_info(job_data)

Create src code info.

参数:

job_data -- job data

返回:

src_code_info

wy_qcos.engine.job_engine.update_src_code_info(src_code_info, aggregation_info)

Update src code info.

参数:

  • src_code_info -- src_code_info

  • aggregation_info -- aggregation info

返回:

src_code_info

wy_qcos.engine.job_engine.get_src_code_cnt(src_code_info)

Get total src code count.

参数:

src_code_info (SourceCodeInfo) -- src_code_info

返回:

src_code_cnt

wy_qcos.engine.job_engine.split_dict(orig_dict, split_len)

Split dict.

参数:

  • orig_dict -- orig_dict

  • split_len -- split_len

返回:

measure_results

wy_qcos.engine.job_engine.get_internal_aggregated_results(job_results, mapping_dict)

Get internal aggregated results.

参数:

  • job_results -- job results

  • mapping_dict -- mapping dict

返回:

aggregated_results

wy_qcos.engine.job_engine.get_external_aggregated_results(job_results, mapping_dict)

Get external aggregated results.

参数:

  • job_results -- job results

  • mapping_dict -- mapping dict

返回:

new job results

wy_qcos.engine.job_engine.job_flow(job_info)

Job flow.

Detail of job flow:
Create task_monitor -> Handle Circuit-Aggregation ->
loop src_code_list ->
[
    run_code ->
        init_driver ->
            driver.validate_driver_configs(device_configs)
            driver.set_configs(device_configs)
            driver.init_driver() ->
            driver.fetch_configs() ->
        init_transpiler ->
        flow_parse ->
            transpiler.parse() ->
        flow_transpile ->
            transpiler.transpile() ->
        flow_run_driver ->
            driver_run ->
                driver.run() / driver.dry_run() ->
    get_results
]
return job_results_list
参数:

job_info -- job info

返回:

results

wy_qcos.engine.job_engine.run_code(source_code_index, src_code_dict, job_info, driver, transpiler, monitor_info)

Run code.

参数:

  • source_code_index -- source code index

  • src_code_dict -- src code dictionary

  • job_info -- job info

  • driver -- driver

  • transpiler -- transpiler

  • monitor_info -- monitor info

返回:

job results

wy_qcos.engine.job_engine.run_qubo_code(source_code_index, src_code_dict, job_info, driver, transpiler)

Run qubo code.

参数:

  • source_code_index -- source code index

  • src_code_dict -- src code dictionary

  • job_info -- job info

  • driver -- driver

  • transpiler -- transpiler

返回:

job results

wy_qcos.engine.job_engine.run_subqubo_code(max_qubits, total_spins_num, source_code_index, src_code_dict, job_info, driver, transpiler)
wy_qcos.engine.job_engine.run_circuit_code(source_code_index, src_code_dict, job_info, driver, transpiler)

Run circuit code.

参数:

  • source_code_index -- source code index

  • src_code_dict -- src code dictionary

  • job_info -- job info

  • driver -- driver

  • transpiler -- transpiler

返回:

job results

wy_qcos.engine.job_engine.run_circuit_cutting_code(source_code_index, src_code_dict, num_qubits, job_info, driver, transpiler)

Run circuit cutting code.

参数:

  • source_code_index -- source code index

  • src_code_dict -- src code dictionary

  • num_qubits -- number of qubits

  • job_info -- job info

  • driver -- driver

  • transpiler -- transpiler

返回:

job results

wy_qcos.engine.job_engine.counts_to_probs(count_dict)

Convert the quantum state count dictionary into a probability array.

参数:

count_dict (dict[str, int]) -- quantum state count dictionary.

返回:

Probability array sorted in binary order.

返回类型:

np.ndarray

wy_qcos.engine.job_engine.probs_to_dict(prob_array)

Generic probability array to dictionary function.

参数:

prob_array (list) -- Probability list

返回:

Probability dictionary

返回类型:

dict

wy_qcos.engine.job_engine.flow_parse(src_code_dict, transpiler, profiling_types, code_type)

Flow: parse.

参数:

  • src_code_dict -- src_code_dict

  • transpiler -- transpiler

  • profiling_types -- profiling types

  • code_type (str) -- code_type

返回:

results, profiling_time

wy_qcos.engine.job_engine.flow_transpile(parsed_src_code, transpiler, driver, profiling_types)

Flow: transpile.

参数:

  • parsed_src_code -- parsed_src_code

  • transpiler -- transpiler

  • driver -- driver

  • profiling_types -- profiling types

返回:

results, profiling_time

wy_qcos.engine.job_engine.flow_task_monitor(monitor_info)

Flow: task monitor.

参数:

monitor_info -- monitor info

wy_qcos.engine.job_engine.flow_run_driver(job_info, num_qubits, driver, data, profiling_types)

Flow: run driver.

参数:

  • job_info -- job info

  • num_qubits -- number of qubits

  • driver -- driver

  • data -- data

  • profiling_types -- profiling types

返回:

results, profiling_time

wy_qcos.engine.job_engine.format_run_results(driver, job_id, data_index)

Format run results.

参数:

  • driver -- driver

  • job_id -- job id

  • data_index -- data index

返回:

formatted results

wy_qcos.engine.job_engine.format_error_results(driver, err_cls, err_msg)

Format error results.

参数:

  • driver -- driver

  • err_cls -- error class

  • err_msg -- error message

返回:

formatted error results

wy_qcos.engine.job_engine.run_job_callback(context, job_results_list)

Run job_callback.

参数:

  • context -- context

  • job_results_list -- list of job results

Module contents