From 143ba2084d7457b07afdba9b9c19105e0a400508 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=98=9A=E5=AF=85?= <13017899+hefei-wu-yanzu@user.noreply.gitee.com> Date: Fri, 6 Feb 2026 01:51:40 +0800 Subject: [PATCH] =?UTF-8?q?1.=E5=8A=A0=E5=85=A5=E5=A4=A9=E8=A1=8D=E5=8E=9F?= =?UTF-8?q?=E7=94=9F=E9=97=A8=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cqlib_adapter/pennylane_ext/1.py | 12 + .../pennylane_ext/circuit_executor.py | 765 +++++------------- cqlib_adapter/pennylane_ext/compilation.py | 244 ++++++ cqlib_adapter/pennylane_ext/device.py | 100 +-- cqlib_adapter/pennylane_ext/native_gates.py | 155 ++++ cqlib_adapter/qiskit_ext/__init__.py | 2 +- cqlib_adapter/qiskit_ext/job.py | 2 +- cqlib_adapter/qiskit_ext/tianyan_backend.py | 2 +- cqlib_adapter/qiskit_ext/tianyan_provider.py | 2 +- .../{qiskit_ext => utils}/api_client.py | 0 test.py | 6 +- tests/test_pennylane/test_compile.py | 24 + 12 files changed, 673 insertions(+), 641 deletions(-) create mode 100644 cqlib_adapter/pennylane_ext/1.py create mode 100644 cqlib_adapter/pennylane_ext/compilation.py create mode 100644 cqlib_adapter/pennylane_ext/native_gates.py rename cqlib_adapter/{qiskit_ext => utils}/api_client.py (100%) create mode 100644 tests/test_pennylane/test_compile.py diff --git a/cqlib_adapter/pennylane_ext/1.py b/cqlib_adapter/pennylane_ext/1.py new file mode 100644 index 0000000..c6463eb --- /dev/null +++ b/cqlib_adapter/pennylane_ext/1.py @@ -0,0 +1,12 @@ +from cqlib_adapter.utils.api_client import ApiClient + + +a = ApiClient(token = "ZtQYpi6GVW24lrSOpauj16mRCAFrWN/3Et4xJjhn7dg=") + +for i in a.get_backends(): + print(i['id']) + print(i['code']) # 量子计算的代码 + print(i['labels']) + print(i['status']) + + \ No newline at end of file diff --git a/cqlib_adapter/pennylane_ext/circuit_executor.py b/cqlib_adapter/pennylane_ext/circuit_executor.py index 222362e..f21e804 100644 --- a/cqlib_adapter/pennylane_ext/circuit_executor.py +++ b/cqlib_adapter/pennylane_ext/circuit_executor.py @@ -3,20 +3,13 @@ This module provides a sophisticated circuit executor that bridges PennyLane quantum circuits with various backend computation platforms, including local simulators, Tianyan simulators, and Tianyan hardware devices. - -Key Features: - - Support for multiple measurement types: expectation values, probabilities, - statevectors, and samples - - Backend-aware execution with automatic capability validation - - Elegant error handling and comprehensive logging - - Seamless integration with PennyLane's quantum tape system """ import logging from enum import Enum from functools import singledispatchmethod from typing import Any, Dict, List, Optional, Union - +from ..utils.api_client import ApiClient import numpy as np import pennylane as qml import json @@ -25,98 +18,70 @@ from cqlib.mapping import transpile_qcis from cqlib.simulator import StatevectorSimulator from cqlib.utils import qasm2 from pennylane.tape import QuantumScript -from pennylane.io import to_openqasm - class BackendType(Enum): - """Enumeration of supported quantum computation backend types. - - Attributes: - LOCAL_SIMULATOR: Local statevector simulator with full measurement support - TIANYAN_SIMULATOR: Tianyan cloud-based simulator supporting probabilities and samples - TIANYAN_HARDWARE: Physical quantum hardware supporting sample measurements only - """ + """Enumeration of supported quantum computation backend types.""" LOCAL_SIMULATOR = "local" TIANYAN_SIMULATOR = "tianyan_simulator" TIANYAN_HARDWARE = "tianyan_hardware" class CircuitExecutor: - """Executes quantum circuits across different computational backends. - - This class serves as the core execution engine for the PennyLane adapter, - providing a unified interface for running quantum circuits on various - backend platforms while handling measurement-specific transformations - and result processing. - - Args: - device_config: Configuration dictionary containing backend settings. - Required keys: - - machine_name: Backend identifier ('default' for local simulator) - - login_key: Authentication key for Tianyan platforms (if applicable) - - shots: Number of measurement shots (None for statevector simulations) - - wires: Number of qubits in the system - - verbose: Enable verbose logging if True - - Raises: - ConnectionError: If backend connection initialization fails - ValueError: If device configuration is invalid or incomplete - - Example: - >>> config = { - ... 'machine_name': 'default', - ... 'shots': 1000, - ... 'wires': 2, - ... 'verbose': True - ... } - >>> executor = CircuitExecutor(config) - >>> result = executor.execute_circuit(quantum_tape) - """ - - #: Set of Tianyan hardware backend identifiers - TIANYAN_HARDWARE_BACKENDS = { - "tianyan24", "tianyan504", "tianyan176-2", "tianyan176" - } - - #: Set of Tianyan simulator backend identifiers - TIANYAN_SIMULATOR_BACKENDS = { - "tianyan_sw", "tianyan_s", "tianyan_tn", - "tianyan_tnn", "tianyan_sa", "tianyan_swn" - } - + """Executes quantum circuits across different computational backends.""" def __init__(self, device_config: Dict[str, Any]) -> None: - """Initialize the circuit executor with device configuration. - - The initialization process includes: - 1. Setting up logging infrastructure - 2. Determining backend type from configuration - 3. Initializing backend connection (for remote platforms) - 4. Preparing measurement processing components - - Args: - device_config: Dictionary containing all necessary configuration - parameters for backend operation and circuit execution. - - Raises: - ConnectionError: If remote backend connection cannot be established - ValueError: If required configuration parameters are missing - """ + """Initialize the circuit executor with device configuration.""" self.device_config = device_config + self.logger = self._setup_logger() + self.cqlib_backend = None self._execution_count = 0 + + self.TIANYAN_HARDWARE_BACKENDS = set() + self.TIANYAN_SIMULATOR_BACKENDS = set() + + machine_name = self.device_config.get('machine_name', 'default') + + if machine_name != "default": + login_key = self.device_config.get('login_key') + if not login_key: + raise ValueError(f"Login key required for backend: {machine_name}") + + try: + self._init_machine_list(login_key) + except Exception as e: + self.logger.error("Failed to fetch backend list: %s", e) + raise ConnectionError(f"Could not connect to Tianyan API: {e}") from e + self._backend_type = self._determine_backend_type() + self._initialize_backend() + self.logger.info("CircuitExecutor initialized with %s backend", self._backend_type.value) - def _setup_logger(self) -> logging.Logger: - """Configure and return a logger instance for execution tracking. + def _init_machine_list(self, token): + """Fetch backend lists from API.""" + + client = ApiClient(token=token) - Returns: - Configured logger instance with appropriate handlers and formatters. - Log level is set to INFO if verbose mode is enabled in configuration. - """ + backends = client.get_backends() + if not backends: + self.logger.warning("No backends retrieved from API.") + return + + for backend in backends: + code = backend.get('code') + label = backend.get('labels') + + if code: + if label == 1: + self.TIANYAN_HARDWARE_BACKENDS.add(code) + else: + self.TIANYAN_SIMULATOR_BACKENDS.add(code) + + def _setup_logger(self) -> logging.Logger: + """Configure and return a logger instance for execution tracking.""" logger = logging.getLogger(f"CircuitExecutor.{id(self)}") if self.device_config.get('verbose', False): @@ -131,14 +96,7 @@ class CircuitExecutor: return logger def _determine_backend_type(self) -> BackendType: - """Determine the appropriate backend type from device configuration. - - Returns: - BackendType enum value corresponding to the configured machine. - - Raises: - ValueError: If machine_name is not recognized or supported - """ + """Determine the appropriate backend type from device configuration.""" backend_name = self.device_config.get('machine_name', 'default') if backend_name == "default": @@ -151,15 +109,7 @@ class CircuitExecutor: raise ValueError(f"Unknown or unsupported backend: {backend_name}") def _initialize_backend(self) -> None: - """Initialize connection to the quantum computation backend. - - For local simulators, no connection is needed. For Tianyan platforms, - this method establishes the API connection using provided credentials. - - Raises: - ConnectionError: If remote backend connection fails - ValueError: If required login credentials are missing - """ + """Initialize connection to the quantum computation backend.""" if self._backend_type == BackendType.LOCAL_SIMULATOR: self.logger.debug("Using local simulator - no backend connection needed") return @@ -181,35 +131,7 @@ class CircuitExecutor: raise ConnectionError(f"Backend connection failed: {error}") from error def execute_circuit(self, circuit: QuantumScript) -> Union[List, Any]: - """Execute a quantum circuit and return measurement results. - - This is the main entry point for circuit execution. It handles the complete - workflow including circuit validation, measurement processing, backend - execution, and result formatting. - - Args: - circuit: PennyLane QuantumScript object containing quantum operations - and measurements to execute. - - Returns: - Single measurement result if the circuit contains only one measurement, - otherwise a list of results corresponding to each measurement in the circuit. - - Raises: - ValueError: If circuit validation fails or backend doesn't support - requested measurements - NotImplementedError: For unsupported measurement types - ConnectionError: If backend execution fails - - Example: - >>> # Circuit with single measurement - >>> result = executor.execute_circuit(tape) - >>> print(f"Expectation value: {result}") - - >>> # Circuit with multiple measurements - >>> results = executor.execute_circuit(tape) - >>> prob_result, sample_result = results - """ + """Execute a quantum circuit and return measurement results.""" self._validate_circuit(circuit) self._execution_count += 1 wire_labels = list(circuit.wires.labels) @@ -224,8 +146,8 @@ class CircuitExecutor: raw_result = self._execute_on_backend(cqlib_circuit, cqlib_qcis) reordered_result = self._reorder_raw_result_by_wire_labels(raw_result, wire_labels) - # Execute with measurement-specific handler + # Execute with measurement-specific handler result = self._execute_measurement(measurement, reordered_result) results.append(result) @@ -234,19 +156,7 @@ class CircuitExecutor: return results[0] if len(results) == 1 else results def _validate_circuit(self, circuit: QuantumScript) -> None: - """Validate circuit constraints and configuration compatibility. - - Ensures that the circuit configuration is compatible with the selected - backend and measurement types. - - Args: - circuit: Quantum circuit to validate - - Raises: - ValueError: If circuit contains state measurements with finite shots - or other incompatible configurations - """ - # Check for state measurement with finite shots + """Validate circuit constraints and configuration compatibility.""" has_state_measurement = any( isinstance(m, qml.measurements.StateMP) for m in circuit.measurements ) @@ -259,31 +169,23 @@ class CircuitExecutor: ) def _validate_measurement_support(self, measurement: Any) -> None: - """Validate that the current backend supports the requested measurement type. - - Args: - measurement: Measurement object to validate - - Raises: - ValueError: If the backend doesn't support the measurement type - """ - # Define measurement support matrix for each backend type + """Validate that the current backend supports the requested measurement type.""" supported_measurements = { BackendType.LOCAL_SIMULATOR: { - qml.measurements.ProbabilityMP, # Probability distributions - qml.measurements.ExpectationMP, # Expectation values - qml.measurements.StateMP, # Full statevector - qml.measurements.SampleMP # Measurement samples + qml.measurements.ProbabilityMP, + qml.measurements.ExpectationMP, + qml.measurements.StateMP, + qml.measurements.SampleMP }, BackendType.TIANYAN_SIMULATOR: { - qml.measurements.ProbabilityMP, # Probability distributions - qml.measurements.SampleMP, # Measurement samples - qml.measurements.ExpectationMP, # Expectation values + qml.measurements.ProbabilityMP, + qml.measurements.SampleMP, + qml.measurements.ExpectationMP, }, BackendType.TIANYAN_HARDWARE: { - qml.measurements.ProbabilityMP, # Probability distributions - qml.measurements.SampleMP, # Measurement samples - qml.measurements.ExpectationMP, # Expectation values + qml.measurements.ProbabilityMP, + qml.measurements.SampleMP, + qml.measurements.ExpectationMP, } } @@ -298,48 +200,104 @@ class CircuitExecutor: def _convert_to_cqlib_format(self, circuit: QuantumScript) -> tuple[Any, str]: """Convert PennyLane circuit to CQLib-compatible format. - Args: - circuit: PennyLane QuantumScript to convert - - Returns: - Tuple containing (cqlib_circuit_object, qcis_instruction_string) - - Raises: - ValueError: If circuit conversion fails + Uses manual QASM construction to support custom native gates that + PennyLane's standard QASM exporter might not handle correctly. """ try: - qasm_string = circuit.to_openqasm() + # Manually build QASM string to include custom gates (x2p, rxy, etc.) + qasm_string = self._build_custom_qasm(circuit) + + # Parse using CQLib's QASM parser + # We assume cqlib.utils.qasm2 supports extended syntax for these native gates cqlib_circuit = qasm2.loads(qasm_string) return cqlib_circuit, cqlib_circuit.qcis except Exception as error: self.logger.error("Circuit conversion from PennyLane to CQLib format failed: %s", error) raise ValueError(f"Circuit conversion failed: {error}") from error - def _execute_measurement(self, measurement: Any, raw_result: Dict[str, Any]) -> Any: - """Execute circuit with measurement-specific processing. - - Dispatches to appropriate measurement handler based on measurement type - using Python's singledispatchmethod for clean, extensible design. - - Args: - measurement: Measurement object defining what to measure - raw_result: Raw result dictionary from backend execution (already reordered) + def _build_custom_qasm(self, circuit: QuantumScript) -> str: + """Build OpenQASM string manually to support custom native gates.""" + num_wires = len(circuit.wires) + + # Standard QASM Header + qasm_lines = ["OPENQASM 2.0;", + "include \"qelib1.inc\";", + f"qreg q[{num_wires}];", + f"creg c[{num_wires}];" + ] + + # Iterate over all operations in the PennyLane Tape + for op in circuit.operations: + op_name = op.name + wires = op.wires.tolist() + params = op.parameters + + # Map wires to qreg indices + q_str = ",".join([f"q[{w}]" for w in wires]) + + # === Custom Native Gates Mapping === + if op_name == "X2PGate": + qasm_lines.append(f"x2p {q_str};") + elif op_name == "X2MGate": + qasm_lines.append(f"x2m {q_str};") + elif op_name == "Y2PGate": + qasm_lines.append(f"y2p {q_str};") + elif op_name == "Y2MGate": + qasm_lines.append(f"y2m {q_str};") + elif op_name == "XY2PGate": + # Parameterized: xy2p(theta) q[0]; + qasm_lines.append(f"xy2p({params[0]}) {q_str};") + elif op_name == "XY2MGate": + qasm_lines.append(f"xy2m({params[0]}) {q_str};") + elif op_name == "RxyGate": + # Two parameters: rxy(phi, theta) q[0]; + qasm_lines.append(f"rxy({params[0]},{params[1]}) {q_str};") + + # === Standard Gates Mapping === + elif op_name == "PauliX": + qasm_lines.append(f"x {q_str};") + elif op_name == "PauliY": + qasm_lines.append(f"y {q_str};") + elif op_name == "PauliZ": + qasm_lines.append(f"z {q_str};") + elif op_name == "Hadamard": + qasm_lines.append(f"h {q_str};") + elif op_name == "RX": + qasm_lines.append(f"rx({params[0]}) {q_str};") + elif op_name == "RY": + qasm_lines.append(f"ry({params[0]}) {q_str};") + elif op_name == "RZ": + qasm_lines.append(f"rz({params[0]}) {q_str};") + elif op_name == "CNOT": + qasm_lines.append(f"cx {q_str};") + elif op_name == "CZ": + qasm_lines.append(f"cz {q_str};") + elif op_name == "S": + qasm_lines.append(f"s {q_str};") + elif op_name == "T": + qasm_lines.append(f"t {q_str};") + else: + # Fallback for other standard gates defined in qelib1.inc + # We try to use PennyLane's own QASM generator for unknown ops + try: + # Note: This might fail if wires don't match the qreg indices exactly 0..N + # but since we initialized mapping, it should be fine. + # Strip the header/footer from standard output + partial_qasm = op.to_openqasm().split('\n') + valid_lines = [l for l in partial_qasm if not l.startswith(('OPENQASM', 'include', 'qreg', 'creg')) and l.strip()] + qasm_lines.extend(valid_lines) + except Exception: + self.logger.warning(f"Operation {op_name} not natively mapped and QASM fallback failed.") + + return "\n".join(qasm_lines) - Returns: - Processed measurement result in PennyLane-compatible format - """ + def _execute_measurement(self, measurement: Any, raw_result: Dict[str, Any]) -> Any: + """Execute circuit with measurement-specific processing.""" return self._execute_measurement_impl(measurement, raw_result) @singledispatchmethod def _execute_measurement_impl(self, measurement: Any, raw_result: Dict[str, Any]) -> Any: - """Base implementation for unsupported measurement types. - - This method is called when no specific handler is registered for - the measurement type. - - Raises: - NotImplementedError: Always raised for unregistered measurement types - """ + """Base implementation for unsupported measurement types.""" raise NotImplementedError( f"Measurement type {type(measurement).__name__} is not supported. " f"Supported types: ProbabilityMP, ExpectationMP, StateMP, SampleMP" @@ -347,60 +305,13 @@ class CircuitExecutor: @_execute_measurement_impl.register def _(self, measurement: qml.measurements.ProbabilityMP, raw_result: Dict[str, Any]) -> np.ndarray: - """Execute probability measurement and return probability distribution. - - Probability measurements return an array where each element represents - the probability of measuring the corresponding computational basis state. - - Args: - measurement: Probability measurement object - raw_result: Raw result dictionary from backend execution - - Returns: - numpy.ndarray: Probability distribution over computational basis states. - The array has length 2^n_qubits and sums to 1.0. - - Example: - >>> # For a 2-qubit system, returns array of length 4 - >>> probs = executor.execute_circuit(tape_with_prob_measurement) - >>> print(f"Probability of |00>: {probs[0]}") - """ + """Execute probability measurement and return probability distribution.""" probabilities = self._extract_probabilities(raw_result) return self._format_probabilities(probabilities) @_execute_measurement_impl.register def _(self, measurement: qml.measurements.ExpectationMP, raw_result: Dict[str, Any]) -> float: - """Execute expectation value measurement for Pauli-Z observables. - - Computes the expectation value of Pauli-Z observables and Pauli-Z strings - directly from probability distribution results. This method assumes the - circuit has already been transformed to the Z-basis measurement. - - Args: - measurement: Expectation measurement object containing a Pauli-Z observable - raw_result: Raw result dictionary containing 'probabilities' key with - a dictionary mapping bitstrings to probabilities - - Returns: - float: Expectation value of the Pauli-Z observable - - Note: - - Only supports Pauli-Z observables and tensor products of Pauli-Z - - Assumes basis transformation has been applied prior to measurement - - Works with probability distributions from both simulators and hardware - - For non-Z observables, use basis transformation in the circuit - - Example: - >>> # Expectation of Z(0) ⊗ Z(1) ⊗ Z(2) - >>> obs = qml.PauliZ(0) @ qml.PauliZ(1) @ qml.PauliZ(2) - >>> expval = executor.execute_circuit(tape_with_expval_measurement) - >>> print(f" = {expval}") - - Raises: - ValueError: If raw_result doesn't contain probability distribution - KeyError: If required keys are missing in raw_result - """ - # Validate input + """Execute expectation value measurement for Pauli-Z observables.""" if 'probabilities' not in raw_result: raise ValueError("raw_result must contain 'probabilities' key") @@ -408,52 +319,26 @@ class CircuitExecutor: if not isinstance(probabilities, dict): raise ValueError("probabilities must be a dictionary") - # Extract qubit indices from observable pauli_indices = list(measurement.obs.wires.labels) - expectation = 0.0 for bitstring, prob in probabilities.items(): - # Calculate eigenvalue for this basis state eigenvalue = 1 for qubit_idx in pauli_indices: if qubit_idx < len(bitstring): - # For Pauli-Z: |0⟩ → +1, |1⟩ → -1 if bitstring[qubit_idx] == '1': eigenvalue *= -1 - - # Expectation = Σ (probability × eigenvalue) expectation += prob * eigenvalue - return expectation @_execute_measurement_impl.register def _(self, measurement: qml.measurements.StateMP, raw_result: Dict[str, Any]) -> np.ndarray: - """Execute statevector measurement and return the full quantum state. - - Returns the complete statevector of the quantum system. This measurement - is only supported on local statevector simulators. - - Args: - measurement: State measurement object - raw_result: Raw result dictionary from backend execution - - Returns: - numpy.ndarray: Complex-valued statevector of length 2^n_qubits - - Raises: - ValueError: If attempted on non-local simulator backend - - Example: - >>> statevector = executor.execute_circuit(tape_with_state_measurement) - >>> print(f"Statevector shape: {statevector.shape}") - """ + """Execute statevector measurement and return the full quantum state.""" if self._backend_type != BackendType.LOCAL_SIMULATOR: raise ValueError( "Statevector measurement is only supported on local simulators. " f"Current backend: {self._backend_type.value}" ) - statevector = raw_result.get('statevector') if statevector is None: raise ValueError("Statevector not found in execution results") @@ -461,81 +346,37 @@ class CircuitExecutor: @_execute_measurement_impl.register def _(self, measurement: qml.measurements.SampleMP, raw_result: Dict[str, Any]) -> np.ndarray: - """Execute sampling measurement and return measurement samples. - - Returns raw measurement samples from multiple circuit executions. - Each sample is a bitstring representing the measurement outcome. - - Args: - measurement: Sample measurement object - raw_result: Raw result dictionary from backend execution - - Returns: - numpy.ndarray: Array of shape (n_shots, n_qubits) where each row - is a measurement outcome and each column is a qubit measurement result. - - Example: - >>> samples = executor.execute_circuit(tape_with_sample_measurement) - >>> print(f"Samples shape: {samples.shape}") - >>> print(f"First measurement: {samples[0]}") - """ + """Execute sampling measurement and return measurement samples.""" samples = raw_result.get('samples') if samples is None: raise ValueError("No measurement samples found in execution results") return samples def _execute_on_backend(self, cqlib_circuit: Any, cqlib_qcis: str) -> Dict[str, Any]: - """Execute circuit on the appropriate backend and return raw results. - - Args: - cqlib_circuit: Circuit in CQLib object format - cqlib_qcis: Circuit in QCIS instruction format - - Returns: - Dictionary containing raw results from backend execution - - Raises: - ConnectionError: If backend execution fails - """ + """Execute circuit on the appropriate backend and return raw results.""" backend_handlers = { BackendType.LOCAL_SIMULATOR: self._execute_local_simulator, BackendType.TIANYAN_SIMULATOR: self._execute_tianyan_simulator, BackendType.TIANYAN_HARDWARE: self._execute_tianyan_hardware } - handler = backend_handlers[self._backend_type] return handler(cqlib_circuit, cqlib_qcis) def _execute_local_simulator(self, cqlib_circuit: Any, cqlib_qcis: str) -> Dict[str, Any]: - """Execute circuit on local statevector simulator. - - Returns comprehensive results including probabilities, samples, and - statevector for local simulation. - - Returns: - Dictionary with keys: 'probabilities', 'samples', 'statevector' - """ + """Execute circuit on local statevector simulator.""" simulator = StatevectorSimulator(cqlib_circuit) - nwe = dict(reversed(simulator.probs().items())) - - reversed_statevector = {key[::-1]: value for key, value in simulator.statevector().items()} + # Assuming simulator.probs() returns {bitstring: prob} return { 'probabilities': simulator.probs(), 'samples': simulator.sample(is_raw_data=True), - 'statevector': reversed_statevector + 'statevector': dict(reversed(simulator.statevector().items())) # Example fix for endianness } def _execute_tianyan_simulator(self, cqlib_circuit: Any, cqlib_qcis: str) -> Dict[str, Any]: - """Execute circuit on Tianyan cloud simulator. - - Returns probability distributions and measurement samples from - Tianyan's cloud-based simulators. - - Returns: - Dictionary with keys: 'probabilities', 'samples' - """ + """Execute circuit on Tianyan cloud simulator.""" + cqlib_circuit.measure_all() query_id = self.cqlib_backend.submit_experiment( - cqlib_qcis, + cqlib_circuit.qcis, num_shots=self.device_config.get('shots') ) raw_result = self.cqlib_backend.query_experiment(query_id)[0] @@ -546,28 +387,18 @@ class CircuitExecutor: } def _execute_tianyan_hardware(self, cqlib_circuit: Any, cqlib_qcis: str) -> Dict[str, Any]: - """Execute circuit on Tianyan quantum hardware. - - Submits circuit to physical quantum hardware and returns measurement - samples. Hardware execution includes readout calibration and error - mitigation where available. - - Returns: - Dictionary with key: 'samples' - - Note: - Hardware execution may involve queueing and longer execution times - """ + """Execute circuit on Tianyan quantum hardware.""" from .ext_mapping import HardwareMapper + cqlib_circuit.measure_all() mapper = self.device_config.get('mapping', None) if mapper: MAP = HardwareMapper(mapper) compiled_circuit = MAP.map_qcis_code(cqlib_circuit.qcis) - else: compiled_circuit = transpile_qcis(cqlib_qcis, self.cqlib_backend)[0] - compiled_circuit = compiled_circuit.qcis - + # Handle return type differences between mapping and transpile + if hasattr(compiled_circuit, 'qcis'): + compiled_circuit = compiled_circuit.qcis query_id = self.cqlib_backend.submit_experiment( compiled_circuit, @@ -585,278 +416,100 @@ class CircuitExecutor: def _reorder_raw_result_by_wire_labels(self, raw_result: Dict[str, Any], wire_labels: List[int]) -> Dict[str, Any]: - """Reorder raw results to match PennyLane's wire label ordering. - - PennyLane may reorder qubits during compilation, and circuit.wires.labels - indicates the final classical bit ordering. This method reorders raw results - before they are processed by measurement-specific handlers. - - Args: - raw_result: The raw result dictionary from backend execution - wire_labels: List of wire labels from circuit.wires.labels, e.g., [0, 2, 1] - - Returns: - Reordered raw result dictionary matching the wire_labels ordering - """ - - reordered_result = raw_result.copy() - - # Reorder probabilities if present - if 'probabilities' in raw_result and raw_result['probabilities']: - reordered_result['probabilities'] = self._reorder_probability_dict( - raw_result['probabilities'], wire_labels - ) - - # Reorder samples if present - if 'samples' in raw_result and raw_result['samples'] is not None: - sample = self._extract_samples(raw_result) - reordered_result['samples'] = self._reorder_sample_matrix( - sample, wire_labels - ) - - # Reorder statevector if present - if 'statevector' in raw_result and raw_result['statevector'] is not None: - statevector = raw_result['statevector'] - reordered_result['statevector'] = self._reorder_statevector( - statevector, wire_labels - ) - - return reordered_result + """Reorder raw results to match PennyLane's wire label ordering.""" + reordered_result = raw_result.copy() + + if 'probabilities' in raw_result and raw_result['probabilities']: + reordered_result['probabilities'] = self._reorder_probability_dict( + raw_result['probabilities'], wire_labels + ) + + if 'samples' in raw_result and raw_result['samples'] is not None: + sample = self._extract_samples(raw_result) + reordered_result['samples'] = self._reorder_sample_matrix( + sample, wire_labels + ) + + if 'statevector' in raw_result and raw_result['statevector'] is not None: + statevector = raw_result['statevector'] + reordered_result['statevector'] = self._reorder_statevector( + statevector, wire_labels + ) + + return reordered_result def _reorder_probability_dict(self, probabilities: Dict[str, float], wire_labels: List[int]) -> Dict[str, float]: """Reorder probability dictionary based on wire labels mapping.""" reordered_probabilities = {} - for bitstring, probability in probabilities.items(): - # Convert to list and reverse for little-endian to big-endian bits = list(bitstring)[::-1] - - # Create new bit array and apply wire label mapping reordered_bits = ['0'] * len(wire_labels) for new_pos, original_pos in enumerate(wire_labels): reordered_bits[original_pos] = bits[new_pos] - - # Convert back to string format reordered_bitstring = ''.join(reordered_bits[::-1]) reordered_probabilities[reordered_bitstring] = probability - return reordered_probabilities def _reorder_sample_matrix(self, samples: np.ndarray, wire_labels: List[int]) -> np.ndarray: - """Reorder sample matrix based on wire labels. - - Args: - samples: Sample matrix of shape (n_shots, n_qubits) - wire_labels: Desired wire ordering, e.g., [0, 2, 1] - - Returns: - Reordered sample matrix - """ - n_qubits = len(wire_labels) - - # Create the permutation to go from current order to desired order - current_to_desired = [wire_labels.index(i) for i in range(n_qubits)] - - # Reorder columns - return samples[:, current_to_desired] + """Reorder sample matrix based on wire labels.""" + n_qubits = len(wire_labels) + current_to_desired = [wire_labels.index(i) for i in range(n_qubits)] + return samples[:, current_to_desired] def _reorder_statevector(self, statevector: Dict[str, complex], wire_labels: List[int]) -> Dict[str, complex]: - """Reorder statevector dictionary based on wire labels. - - Args: - statevector: Dictionary mapping bitstrings to complex amplitudes - wire_labels: Desired wire ordering, e.g., [0, 2, 1] - - Returns: - Reordered statevector dictionary - """ + """Reorder statevector dictionary based on wire labels.""" n_qubits = len(wire_labels) - - # Create the permutation to go from current order to desired order - # For example, if wire_labels = [0, 2, 1], then: - # current_to_desired = [0, 2, 1] means: - # - bit 0 stays at position 0 - # - bit 1 goes to position 2 - # - bit 2 goes to position 1 current_to_desired = [wire_labels.index(i) for i in range(n_qubits)] - reordered_statevector = {} - for bitstring, amplitude in statevector.items(): - # Convert bitstring to list of bits bits = list(bitstring) - # Reorder bits according to the permutation reordered_bits = [bits[current_to_desired[i]] for i in range(n_qubits)] reordered_bitstring = ''.join(reordered_bits) - - # Store with reordered bitstring reordered_statevector[reordered_bitstring] = amplitude - return reordered_statevector def _extract_probabilities(self, raw_result: Dict[str, Any]) -> Dict[str, float]: - """Extract and process probability distribution from raw results. - - Handles endianness conversion to ensure consistent little-endian - format across all backends. - - Args: - raw_result: Raw result dictionary from backend execution - - Returns: - Probability dictionary mapping bitstrings to probabilities - """ + """Extract and process probability distribution from raw results.""" probabilities = raw_result.get('probabilities', {}) - - # Convert to little-endian format for consistency if probabilities and isinstance(probabilities, dict): return {key[::-1]: value for key, value in probabilities.items()} - return probabilities def _extract_samples(self, raw_result: Dict[str, Any]) -> Any: - """Extract and format samples from raw results. - - Converts local simulator samples to PennyLane-compatible format - while preserving Tianyan backend sample formats. - - Args: - raw_result: Raw result dictionary from backend execution - - Returns: - Formatted samples appropriate for the backend type - """ + """Extract and format samples from raw results.""" samples = raw_result.get('samples') - - # Convert local simulator samples to standard format if samples is not None and self._backend_type == BackendType.LOCAL_SIMULATOR: return samples_to_pennylane_format(samples, self.device_config['wires']) - return samples def _format_probabilities(self, probabilities: Dict[str, float]) -> np.ndarray: - """Convert probability dictionary to PennyLane array format. - - Args: - probabilities: Dictionary mapping bitstrings to probabilities - - Returns: - numpy.ndarray: Probability array indexed by computational basis states - - Raises: - ValueError: If probability dictionary is empty or invalid - """ + """Convert probability dictionary to PennyLane array format.""" if not probabilities: raise ValueError("No probability distribution found in execution results") - num_qubits = len(next(iter(probabilities.keys()))) prob_array = np.zeros(2 ** num_qubits) - for bitstring, prob in probabilities.items(): - index = int(bitstring, 2) # Convert binary string to integer index + index = int(bitstring, 2) prob_array[index] = prob - return prob_array - def _format_samples(self, samples: Any, measurement: qml.measurements.SampleMP) -> np.ndarray: - """Format samples for PennyLane compatibility. - - Args: - samples: Raw samples from backend execution - measurement: Sample measurement object for context - - Returns: - Formatted samples array - Raises: - ValueError: If no samples are found in results - """ - if samples is None: - raise ValueError("No measurement samples found in execution results") - return samples - - def get_execution_stats(self) -> Dict[str, Any]: - """Get execution statistics and performance metrics. - - Returns: - Dictionary containing execution count, backend information, - and configuration details. - - Example: - >>> stats = executor.get_execution_stats() - >>> print(f"Total executions: {stats['execution_count']}") - >>> print(f"Backend type: {stats['backend_type']}") - """ - return { - "execution_count": self._execution_count, - "backend_type": self._backend_type.value, - "wires": self.device_config.get('wires'), - "shots": self.device_config.get('shots'), - "machine_name": self.device_config.get('machine_name') - } - - -def decimal_to_binary_array( - decimal_value: int, - num_bits: int, - little_endian: bool = True -) -> np.ndarray: - """Convert decimal integer to binary array representation. - - Args: - decimal_value: Integer value to convert to binary - num_bits: Number of bits in the binary representation - little_endian: If True, least significant bit is at index 0. - If False, most significant bit is at index 0. - - Returns: - numpy.ndarray: Binary array of length num_bits containing 0s and 1s - - Example: - >>> decimal_to_binary_array(5, 4, little_endian=True) - array([1, 0, 1, 0]) # 5 = 1*2^0 + 0*2^1 + 1*2^2 + 0*2^3 - >>> decimal_to_binary_array(5, 4, little_endian=False) - array([0, 1, 0, 1]) # 5 = 0*2^3 + 1*2^2 + 0*2^1 + 1*2^0 - """ +# Helper functions (kept as provided) +def decimal_to_binary_array(decimal_value: int, num_bits: int, little_endian: bool = True) -> np.ndarray: binary_string = np.binary_repr(int(decimal_value), width=num_bits) bits = np.array([int(bit) for bit in binary_string]) return bits[::-1] if little_endian else bits - def samples_to_pennylane_format( samples: Union[List[int], np.ndarray], num_qubits: Optional[int] = None, measured_qubits: Optional[List[int]] = None, little_endian: bool = True ) -> np.ndarray: - """Convert decimal samples to PennyLane-compatible binary matrix. - - Args: - samples: Array of decimal integers representing measurement outcomes - num_qubits: Total number of qubits in the system - measured_qubits: Specific qubits that were measured - little_endian: Endianness convention for bit ordering - - Returns: - numpy.ndarray: Binary matrix of shape (n_shots, n_bits) where - each row is a measurement outcome and each column is a qubit result. - - Raises: - ValueError: If number of bits cannot be determined from inputs - - Example: - >>> samples = [1, 3, 2] # Decimal measurement outcomes - >>> samples_to_pennylane_format(samples, num_qubits=2) - array([[1, 0], # 1 in binary (little-endian) - [1, 1], # 3 in binary - [0, 1]]) # 2 in binary - """ samples_array = np.asarray(samples) - - # Determine required number of bits if measured_qubits is not None: num_bits = len(measured_qubits) elif num_qubits is not None: @@ -867,34 +520,8 @@ def samples_to_pennylane_format( max_value = np.max(samples_array) num_bits = int(np.ceil(np.log2(max_value + 1))) if max_value > 0 else 1 - # Convert each sample to binary representation n_shots = len(samples_array) binary_matrix = np.zeros((n_shots, num_bits), dtype=int) - for i, sample in enumerate(samples_array): binary_matrix[i] = decimal_to_binary_array(sample, num_bits, little_endian) - - return binary_matrix - - -def switch_endianness( - binary_data: Union[List[int], np.ndarray, List[List[int]]] -) -> np.ndarray: - """Reverse the bit order (endianness) of binary data. - - Args: - binary_data: Binary data to convert. Can be 1D array (single measurement) - or 2D array (multiple measurements, each row is a bitstring) - - Returns: - numpy.ndarray: Binary data with bit order reversed along the last axis - - Example: - >>> switch_endianness([1, 0, 1, 0]) - array([0, 1, 0, 1]) # Big-endian to little-endian - >>> switch_endianness([[1, 0], [0, 1]]) - array([[0, 1], # Each row reversed independently - [1, 0]]) - """ - data_array = np.asarray(binary_data) - return data_array[..., ::-1] # Reverse along the last axis (bit dimension) \ No newline at end of file + return binary_matrix \ No newline at end of file diff --git a/cqlib_adapter/pennylane_ext/compilation.py b/cqlib_adapter/pennylane_ext/compilation.py new file mode 100644 index 0000000..342fca6 --- /dev/null +++ b/cqlib_adapter/pennylane_ext/compilation.py @@ -0,0 +1,244 @@ +"""Quantum circuit compilation module for PennyLane and CQLib. + +This module provides functions to transform standard PennyLane quantum circuits +into native gate sequences compatible with the CQLib hardware backend. +""" + +from typing import Any, List, Union + +import numpy as np +import pennylane as qml +from cqlib import Circuit +from pennylane.tape import QuantumScript +from pennylane.transforms import decompose +from pennylane.workflow import construct_tape + +# Local application imports +from native_gates import ( + X2MGate, + X2PGate, + XY2MGate, + XY2PGate, + Y2MGate, + Y2PGate, +) + +# Constants +PI = np.pi +PI_2 = np.pi / 2 + +def compile_to_native_gates(qnode: Any, *params: Any) -> QuantumScript: + """Decomposes a QNode into a native gate set as a PennyLane QuantumScript. + + Args: + qnode: The PennyLane QNode to be compiled. + *params: Parameters required to execute the QNode. + + Returns: + A QuantumScript containing the compiled native PennyLane gates. + """ + # Construct the tape from the QNode with specific parameters + tape = construct_tape(qnode)(*params) + + # Decompose into a standard base gate set first + batch, _ = decompose( + tape, + gate_set={qml.CNOT, qml.RX, qml.RY, qml.RZ}, + max_expansion=10 + ) + decomposed_tape: QuantumScript = batch[0] + + new_ops = [] + + for op in decomposed_tape.operations: + name = op.name + wires = op.wires + p = op.parameters + + # Mapping logic + if name == "Hadamard": + # Map H -> RZ(pi/2) - X2P - RZ(pi/2) + new_ops.append(qml.RZ(PI_2, wires=wires)) + new_ops.append(X2PGate(wires=wires)) + new_ops.append(qml.RZ(PI_2, wires=wires)) + + elif name == "CNOT": + # Map CNOT -> H(target) - CZ - H(target) + ctrl, target = wires[0], wires[1] + target_wires = [target] + # H on target + new_ops.append(qml.RZ(PI_2, wires=target_wires)) + new_ops.append(X2PGate(wires=target_wires)) + new_ops.append(qml.RZ(PI_2, wires=target_wires)) + # Native CZ + new_ops.append(qml.CZ(wires=[ctrl, target])) + # H on target again + new_ops.append(qml.RZ(PI_2, wires=target_wires)) + new_ops.append(X2PGate(wires=target_wires)) + new_ops.append(qml.RZ(PI_2, wires=target_wires)) + + elif name == "RX": + # Map RX(theta) -> RZ(-pi/2) - X2P - RZ(theta) - X2M - RZ(pi/2) + theta = p[0] + new_ops.append(qml.RZ(-PI_2, wires=wires)) + new_ops.append(X2PGate(wires=wires)) + new_ops.append(qml.RZ(theta, wires=wires)) + new_ops.append(X2MGate(wires=wires)) + new_ops.append(qml.RZ(PI_2, wires=wires)) + + elif name == "RY": + # Map RY(theta) -> X2P - RZ(theta) - X2M + theta = p[0] + new_ops.append(X2PGate(wires=wires)) + new_ops.append(qml.RZ(theta, wires=wires)) + new_ops.append(X2MGate(wires=wires)) + + elif name == "PauliX": + # Map X -> X2P - X2P + new_ops.append(X2PGate(wires=wires)) + new_ops.append(X2PGate(wires=wires)) + + elif name == "PauliY": + # Map Y -> Y2P - Y2P + new_ops.append(Y2PGate(wires=wires)) + new_ops.append(Y2PGate(wires=wires)) + + elif name in ["RZ", "CZ", "Barrier"]: + # Retain standard gates that are native + new_ops.append(op) + + elif name in [ + "X2PGate", "X2MGate", "Y2PGate", "Y2MGate", "XY2PGate", "XY2MGate" + ]: + # Retain custom native gates + new_ops.append(op) + + else: + print(f"Warning: Operation {name} not natively mapped. Kept.") + new_ops.append(op) + + return QuantumScript(ops=new_ops, measurements=decomposed_tape.measurements) + + +def compile_to_native_cqlib(qnode: Any, *params: Any) -> Circuit: + """Reconstructs a PennyLane circuit into a strict CQLib native circuit. + + Args: + qnode: The PennyLane QNode to be compiled. + *params: Parameters required to execute the QNode. + + Returns: + A cqlib.Circuit object containing strictly native instructions. + """ + # 1. Get decomposed Tape + raw_tape = construct_tape(qnode)(*params) + base_gate_set = { + qml.RX, qml.RY, qml.RZ, qml.CNOT, qml.CZ, qml.Hadamard, + qml.PauliX, qml.PauliY, qml.PauliZ, qml.S, qml.T + } + batch, _ = decompose(raw_tape, gate_set=base_gate_set, max_expansion=10) + tape: QuantumScript = batch[0] + + # 2. Initialize CQLib circuit + cql_circ = Circuit(tape.num_wires) + + # 3. Explicit manual mapping to physical native gates + for op in tape.operations: + name = op.name + w = op.wires.tolist() + p = op.parameters + + if name == "Hadamard": + # H Q1 -> RZ Q1 PI, Y2P Q1 + cql_circ.rz(w[0], PI) + cql_circ.y2p(w[0]) + + elif name == "CNOT": + # CX Q0 Q1 -> Y2M Q1, CZ Q0 Q1, Y2P Q1 + cql_circ.y2m(w[1]) + cql_circ.cz(w[0], w[1]) + cql_circ.y2p(w[1]) + + elif name == "RX": + # RX Q1 theta -> RZ Q1 PI_2, X2P Q1, RZ Q1 theta, X2M Q1, RZ Q1 -PI_2 + theta = p[0] + cql_circ.rz(w[0], PI_2) + cql_circ.x2p(w[0]) + cql_circ.rz(w[0], theta) + cql_circ.x2m(w[0]) + cql_circ.rz(w[0], -PI_2) + + elif name == "RY": + # RY Q1 theta -> X2P Q1, RZ Q1 theta, X2M Q1 + theta = p[0] + cql_circ.x2p(w[0]) + cql_circ.rz(w[0], theta) + cql_circ.x2m(w[0]) + + elif name == "PauliX": + # X Q1 -> X2P Q1, X2P Q1 + cql_circ.x2p(w[0]) + cql_circ.x2p(w[0]) + + elif name == "PauliY": + # Y Q1 -> Y2P Q1, Y2P Q1 + cql_circ.y2p(w[0]) + cql_circ.y2p(w[0]) + + elif name == "PauliZ": + cql_circ.rz(w[0], PI) + + elif name == "S": + cql_circ.rz(w[0], PI_2) + + elif name == "T": + cql_circ.rz(w[0], PI / 4) + + elif name == "RZ": + cql_circ.rz(w[0], p[0]) + + elif name == "CZ": + cql_circ.cz(w[0], w[1]) + + elif name == "Barrier": + cql_circ.barrier(*w) + + elif name == "Identity": + cql_circ.i(w[0], 0) + + # Handling custom native gates + elif name == "X2PGate": + cql_circ.x2p(w[0]) + elif name == "X2MGate": + cql_circ.x2m(w[0]) + elif name == "Y2PGate": + cql_circ.y2p(w[0]) + elif name == "Y2MGate": + cql_circ.y2m(w[0]) + elif name == "XY2PGate": + cql_circ.xy2p(w[0], p[0]) + elif name == "XY2MGate": + cql_circ.xy2m(w[0], p[0]) + + else: + print(f"Warning: Operation {name} not mapped to native CQLib.") + + # 4. Final measurement processing + if tape.measurements: + cql_circ.measure_all() + + return cql_circ + + +if __name__ == "__main__": + # Example execution + + @qml.qnode(qml.device("default.qubit", wires=3)) + def circuit() -> Any: + """Example PennyLane QNode with standard gates.""" + qml.Hadamard(wires=0) + qml.Toffoli(wires=[0, 1, 2]) + return qml.state() + cqlib_circuit = compile_to_native_cqlib(circuit) + print("Compiled QCIS Instructions:") + print(cqlib_circuit.qcis) \ No newline at end of file diff --git a/cqlib_adapter/pennylane_ext/device.py b/cqlib_adapter/pennylane_ext/device.py index 4ecb4af..263963a 100644 --- a/cqlib_adapter/pennylane_ext/device.py +++ b/cqlib_adapter/pennylane_ext/device.py @@ -12,7 +12,11 @@ from pennylane.devices import Device from pennylane.tape import QuantumScript, QuantumScriptOrBatch from .circuit_executor import CircuitExecutor - +from .native_gates import ( + X2PGate, X2MGate, + Y2PGate, Y2MGate, + XY2PGate, XY2MGate +) class CQLibDevice(Device): """Custom quantum device implementing PennyLane device interface using CQLib backend. @@ -46,7 +50,9 @@ class CQLibDevice(Device): } # Supported operations + # Updated to include custom native gates (X2P, Y2P, Rxy, etc.) SUPPORTED_OPERATIONS = { + # Standard Gates "Hadamard", "PauliX", "PauliY", @@ -56,6 +62,15 @@ class CQLibDevice(Device): "RX", "RY", "RZ", + "S", + "T", + + "X2PGate", + "X2MGate", + "Y2PGate", + "Y2MGate", + "XY2PGate", + "XY2MGate", } # Device metadata @@ -101,29 +116,17 @@ class CQLibDevice(Device): @property def name(self) -> str: - """Return the device name. - - Returns: - String representing the device name. - """ + """Return the device name.""" return "Cqlib Quantum Device" @property def operations(self) -> Set[str]: - """Return the set of supported operations. - - Returns: - Set of supported operation names. - """ + """Return the set of supported operations.""" return self.SUPPORTED_OPERATIONS @property def backend_info(self) -> Dict[str, Any]: - """Return information about the current backend. - - Returns: - Dictionary containing backend configuration information. - """ + """Return information about the current backend.""" return { "backend_type": self.machine_name, "is_hardware": self.machine_name in self.TIANYAN_HARDWARE_BACKENDS, @@ -134,11 +137,7 @@ class CQLibDevice(Device): @classmethod def capabilities(cls) -> Dict[str, Any]: - """Return the device capabilities configuration. - - Returns: - Dictionary containing supported features and capabilities. - """ + """Return the device capabilities configuration.""" capabilities = super().capabilities().copy() capabilities.update( @@ -154,70 +153,32 @@ class CQLibDevice(Device): "jax": "default.qubit.jax", }, ) - return capabilities def supports_operation(self, operation: Any) -> bool: """Check if a specific quantum operation is supported. - Args: - operation: Quantum operation to check. - - Returns: - True if the operation is supported, False otherwise. + This method is critical for preventing PennyLane from decomposing + our native gates (like X2PGate) into standard gates. """ - supported_operations = { - "PauliX", - "PauliY", - "PauliZ", - "Hadamard", - "S", - "T", - "RX", - "RY", - "RZ", - "CNOT", - "CZ", - } - return getattr(operation, "name", None) in supported_operations + return getattr(operation, "name", None) in self.SUPPORTED_OPERATIONS def execute( self, circuits: Union[QuantumScript, List[QuantumScript]], execution_config: Any = None, ) -> List[Any]: - """Execute quantum circuits on the device. - - Args: - circuits: Single quantum circuit or list of circuits to execute. - execution_config: Execution configuration parameters. - - Returns: - List of execution results for each circuit. - """ + """Execute quantum circuits on the device.""" if isinstance(circuits, QuantumScript): circuits = [circuits] return [self.circuit_executor.execute_circuit(circuit) for circuit in circuits] - def __repr__(self) -> str: - """Return string representation of the device. - - Returns: - String representation of the device. - """ return f"<{self.name} device (wires={self.wires}, shots={self.shots})>" def preprocess_transforms(self, execution_config: Any = None) -> Any: - """Define the preprocessing transformation pipeline. - - Args: - execution_config: Execution configuration parameters. - - Returns: - TransformProgram: Preprocessing transformation program. - """ + """Define the preprocessing transformation pipeline.""" program = qml.transforms.core.TransformProgram() program.add_transform( qml.devices.preprocess.validate_device_wires, @@ -228,9 +189,18 @@ class CQLibDevice(Device): qml.devices.preprocess.validate_measurements, name=self.short_name, ) + + # IMPORTANT: stopping_condition uses self.supports_operation + # to preserve our custom native gates. program.add_transform( qml.devices.preprocess.decompose, stopping_condition=self.supports_operation, name=self.short_name, ) - return program \ No newline at end of file + return program + + + + + + diff --git a/cqlib_adapter/pennylane_ext/native_gates.py b/cqlib_adapter/pennylane_ext/native_gates.py new file mode 100644 index 0000000..19103e0 --- /dev/null +++ b/cqlib_adapter/pennylane_ext/native_gates.py @@ -0,0 +1,155 @@ +# gates/native_gates.py +import pennylane as qml +from pennylane.operation import Operation +from pennylane.typing import TensorLike +from pennylane.wires import WiresLike +import numpy as np + +def _stack_matrix(elements): + return qml.math.stack([ + qml.math.stack(elements[:2], axis=-1), + qml.math.stack(elements[2:], axis=-1) + ], axis=-2) + + +class X2PGate(Operation): + r"""Native X rotation by +pi/2 (Sqrt X).""" + num_wires = 1 + num_params = 0 + grad_method = None + + @staticmethod + def compute_matrix() -> TensorLike: + # 1/sqrt(2) * [[1, -i], [-i, 1]] + isqrt2 = 1 / np.sqrt(2) + return qml.math.array([ + [isqrt2, -1j * isqrt2], + [-1j * isqrt2, isqrt2] + ], dtype=complex) + + def adjoint(self): + return X2MGate(wires=self.wires) + +class X2MGate(Operation): + r"""Native X rotation by -pi/2 (Sqrt X dag).""" + num_wires = 1 + num_params = 0 + grad_method = None + + @staticmethod + def compute_matrix() -> TensorLike: + isqrt2 = 1 / np.sqrt(2) + return qml.math.array([ + [isqrt2, 1j * isqrt2], + [1j * isqrt2, isqrt2] + ], dtype=complex) + + def adjoint(self): + return X2PGate(wires=self.wires) + +class Y2PGate(Operation): + r"""Native Y rotation by +pi/2 (Sqrt Y).""" + num_wires = 1 + num_params = 0 + grad_method = None + + @staticmethod + def compute_matrix() -> TensorLike: + isqrt2 = 1 / np.sqrt(2) + return qml.math.array([ + [isqrt2, -1 * isqrt2], + [1 * isqrt2, isqrt2] + ], dtype=complex) + + def adjoint(self): + return Y2MGate(wires=self.wires) + +class Y2MGate(Operation): + r"""Native Y rotation by -pi/2 (Sqrt Y dag).""" + num_wires = 1 + num_params = 0 + grad_method = None + + @staticmethod + def compute_matrix() -> TensorLike: + isqrt2 = 1 / np.sqrt(2) + return qml.math.array([ + [isqrt2, 1 * isqrt2], + [-1 * isqrt2, isqrt2] + ], dtype=complex) + + def adjoint(self): + return Y2PGate(wires=self.wires) + +# --- 参数化门 (Parameterized Gates) --- + +class XY2PGate(Operation): + r""" + Rotation in XY plane by +pi/2. + Axis defined by phi. + """ + num_wires = 1 + num_params = 1 + ndim_params = (0,) + grad_method = "A" # 支持自动微分 + + def __init__(self, phi: TensorLike, wires: WiresLike, id=None): + super().__init__(phi, wires=wires, id=id) + + @staticmethod + def compute_matrix(phi) -> TensorLike: + if qml.math.get_interface(phi) == "tensorflow": + phi = qml.math.cast_like(phi, 1j) + + c = qml.math.cos(phi) + s = qml.math.sin(phi) + + one = qml.math.cast_like(qml.math.ones_like(phi), 1j) + neg_i = -1j * one + + # -i * e^{-i*phi} + term1 = neg_i * (c - 1j * s) + # -i * e^{i*phi} + term2 = neg_i * (c + 1j * s) + + isqrt2 = 1 / np.sqrt(2) + mat = _stack_matrix([one, term1, term2, one]) + return mat * isqrt2 + + def adjoint(self): + return XY2MGate(self.data[0], wires=self.wires) + +class XY2MGate(Operation): + r""" + Rotation in XY plane by -pi/2. + """ + num_wires = 1 + num_params = 1 + ndim_params = (0,) + grad_method = "A" + + def __init__(self, phi: TensorLike, wires: WiresLike, id=None): + super().__init__(phi, wires=wires, id=id) + + @staticmethod + def compute_matrix(phi) -> TensorLike: + if qml.math.get_interface(phi) == "tensorflow": + phi = qml.math.cast_like(phi, 1j) + + c = qml.math.cos(phi) + s = qml.math.sin(phi) + + one = qml.math.cast_like(qml.math.ones_like(phi), 1j) + pos_i = 1j * one + + # +i * e^{-i*phi} + term1 = pos_i * (c - 1j * s) + # +i * e^{i*phi} + term2 = pos_i * (c + 1j * s) + + isqrt2 = 1 / np.sqrt(2) + mat = _stack_matrix([one, term1, term2, one]) + return mat * isqrt2 + + def adjoint(self): + return XY2PGate(self.data[0], wires=self.wires) \ No newline at end of file diff --git a/cqlib_adapter/qiskit_ext/__init__.py b/cqlib_adapter/qiskit_ext/__init__.py index 21fe9f8..019546b 100644 --- a/cqlib_adapter/qiskit_ext/__init__.py +++ b/cqlib_adapter/qiskit_ext/__init__.py @@ -15,7 +15,7 @@ Qiskit extension module for TianYan quantum platform integration. """ from .adapter import to_cqlib -from .api_client import ApiClient +from ..utils.api_client import ApiClient from .gates import X2PGate, X2MGate, Y2PGate, Y2MGate, XY2MGate, XY2PGate, RxyGate from .job import TianYanJob from .sampler import TianYanSampler diff --git a/cqlib_adapter/qiskit_ext/job.py b/cqlib_adapter/qiskit_ext/job.py index 31fad98..ffdc834 100644 --- a/cqlib_adapter/qiskit_ext/job.py +++ b/cqlib_adapter/qiskit_ext/job.py @@ -26,7 +26,7 @@ from qiskit.providers import JobV1, JobStatus from qiskit.providers.backend import Backend from qiskit.result import Result -from .api_client import ApiClient +from ..utils.api_client import ApiClient class TianYanJob(JobV1): diff --git a/cqlib_adapter/qiskit_ext/tianyan_backend.py b/cqlib_adapter/qiskit_ext/tianyan_backend.py index 4336625..ae889f7 100644 --- a/cqlib_adapter/qiskit_ext/tianyan_backend.py +++ b/cqlib_adapter/qiskit_ext/tianyan_backend.py @@ -31,7 +31,7 @@ from qiskit.providers import BackendV2 as Backend, Options, JobV1, QubitProperti from qiskit.transpiler import Target, InstructionProperties, generate_preset_pass_manager from .adapter import to_cqlib -from .api_client import ApiClient +from ..utils.api_client import ApiClient from .gates import X2PGate, X2MGate, Y2MGate, Y2PGate, XY2MGate, XY2PGate, RxyGate from .job import TianYanJob diff --git a/cqlib_adapter/qiskit_ext/tianyan_provider.py b/cqlib_adapter/qiskit_ext/tianyan_provider.py index 916a736..7660e8d 100644 --- a/cqlib_adapter/qiskit_ext/tianyan_provider.py +++ b/cqlib_adapter/qiskit_ext/tianyan_provider.py @@ -22,7 +22,7 @@ from pathlib import Path import dotenv -from .api_client import ApiClient +from ..utils.api_client import ApiClient from .tianyan_backend import TianYanBackend, BackendConfiguration, CqlibAdapterError, \ BackendStatus, TianYanQuantumBackend, TianYanSimulatorBackend diff --git a/cqlib_adapter/qiskit_ext/api_client.py b/cqlib_adapter/utils/api_client.py similarity index 100% rename from cqlib_adapter/qiskit_ext/api_client.py rename to cqlib_adapter/utils/api_client.py diff --git a/test.py b/test.py index 44845b9..6df03ce 100644 --- a/test.py +++ b/test.py @@ -16,15 +16,15 @@ from pennylane.tape import QuantumScript, QuantumScriptOrBatch # 创建设备 # dev = qml.device("cqlib.device",wires=[0,1,2]) -dev = qml.device("default.qubit",wires=[0,1,2],shots = 1000) +dev = qml.device("cqlib.device",wires=3,shots = 1000,login_key = "ZtQYpi6GVW24lrSOpauj16mRCAFrWN/3Et4xJjhn7dg=",cqlib_backend_name = 'tianyan_sw') @qml.qnode(dev, interface='auto',autograph = False) def circuit(): qml.H(0) qml.CNOT(wires=[0, 2]) # 计算 Z⊗Z⊗Z 的期望值 # return qml.expval(qml.PauliX(0) @ qml.PauliZ(1) @ qml.PauliZ(2)) - # return qml.probs() - return qml.sample() + return qml.probs() + # 执行电路 result = circuit() diff --git a/tests/test_pennylane/test_compile.py b/tests/test_pennylane/test_compile.py new file mode 100644 index 0000000..02f7018 --- /dev/null +++ b/tests/test_pennylane/test_compile.py @@ -0,0 +1,24 @@ +"""Quantum Circuit Executor for PennyLane Adapter. + +This module provides a sophisticated circuit executor that bridges PennyLane quantum +circuits with various backend computation platforms, including local simulators, +Tianyan simulators, and Tianyan hardware devices. +""" + +import logging +from enum import Enum +from functools import singledispatchmethod +from typing import Any, Dict, List, Optional, Union + +import numpy as np +import pennylane as qml +import json +from cqlib import TianYanPlatform +from cqlib.mapping import transpile_qcis +from cqlib.simulator import StatevectorSimulator +from cqlib.utils import qasm2 +from pennylane.tape import QuantumScript + + + + -- Gitee