GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/qdk

Public

mirrored from https://github.com/microsoft/qdkAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
copilot/replace-qsharp-with-qdk-python-tests

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

qdk/source/pip/qsharp/interop/qiskit/backends

source/pip/qsharp/interop/qiskit/backends/backend_base.py

614lines · modecode

RawDownload
Latest commit unavailable.
1# Copyright (c) Microsoft Corporation.
2# Licensed under the MIT License.
3
4from abc import ABC, abstractmethod
5import datetime
6import logging
7from time import monotonic
8from typing import Dict, Any, List, Optional, Union
9from warnings import warn
10
11from qiskit import transpile
12from qiskit.circuit import (
13 QuantumCircuit,
14)
15from qiskit.version import get_version_info
16
17from qiskit.qasm3.exporter import Exporter
18from qiskit.providers import BackendV2, Options
19from qiskit.result import Result
20from qiskit.transpiler import PassManager
21from qiskit.transpiler.passes import RemoveBarriers, RemoveResetInZeroState
22from qiskit.transpiler.target import Target
23
24from .compilation import Compilation
25from .errors import Errors
26from .qirtarget import QirTarget
27from ..execution import DetaultExecutor
28from ..jobs import QsJob, QsSimJob, QsJobSet
29from ..passes import RemoveDelays
30from .... import TargetProfile
31
32logger = logging.getLogger(__name__)
33
34_QISKIT_NON_GATE_INSTRUCTIONS = [
35 "control_flow",
36 "if_else",
37 "switch_case",
38 "while_loop",
39 "break",
40 "continue",
41 "store",
42 "for_loop",
43 "measure",
44 "reset",
45]
46
47_QISKIT_STDGATES = [
48 "p",
49 "x",
50 "y",
51 "z",
52 "h",
53 "s",
54 "sdg",
55 "t",
56 "tdg",
57 "sx",
58 "rx",
59 "ry",
60 "rz",
61 "cx",
62 "cy",
63 "cz",
64 "cp",
65 "crx",
66 "cry",
67 "crz",
68 "ch",
69 "swap",
70 "ccx",
71 "cswap",
72 "cu",
73 "CX",
74 "phase",
75 "cphase",
76 "id",
77 "u1",
78 "u2",
79 "u3",
80 "U",
81]
82
83
84def filter_kwargs(func, **kwargs) -> Dict[str, Any]:
85 import inspect
86
87 sig = inspect.signature(func)
88 supported_args = set(sig.parameters.keys())
89 extracted_kwargs = {
90 k: kwargs.get(k) for k in list(kwargs.keys()) if k in supported_args
91 }
92 return extracted_kwargs
93
94
95def get_transpile_options(**kwargs) -> Dict[str, Any]:
96 args = filter_kwargs(transpile, **kwargs)
97 return args
98
99
100def get_exporter_options(**kwargs) -> Dict[str, Any]:
101 return filter_kwargs(Exporter.__init__, **kwargs)
102
103
104class BackendBase(BackendV2, ABC):
105 """
106 A virtual backend for transpiling to a Q# ecosystem compatible format.
107 """
108
109 def __init__(
110 self,
111 target: Optional[Target] = None,
112 qiskit_pass_options: Optional[Dict[str, Any]] = None,
113 transpile_options: Optional[Dict[str, Any]] = None,
114 qasm_export_options: Optional[Dict[str, Any]] = None,
115 skip_transpilation: bool = False,
116 **options,
117 ):
118 """
119 :param target: The target to use for the backend.
120 :param qiskit_pass_options: Options for the Qiskit passes.
121 :param transpile_options: Options for the transpiler.
122 :param qasm_export_options: Options for the QASM3 exporter.
123 :param **options: Additional keyword arguments passed to subclasses.
124 """
125 super().__init__(
126 name="QSharpBackend",
127 description="A virtual BackendV2 for transpiling to a Q# compatible format.",
128 backend_version="0.0.1",
129 )
130
131 if options is not None:
132 # we need to rename the seed_simulator to seed. This
133 # is a convenience for aer users.
134 # if the user passes in seed_simulator, we will rename it to seed
135 # but only if the seed field is defined in the backend options.
136 if "seed_simulator" in options and "seed" in self._options.data:
137 warn("seed_simulator passed, but field is called seed.")
138 options["seed"] = options.pop("seed_simulator")
139
140 # updates the options with the fields passed in, if the backend
141 # doesn't have the field, it will raise an error.
142 self.set_options(**options)
143
144 self._qiskit_pass_options = Options(
145 supports_barrier=False,
146 supports_delay=False,
147 remove_reset_in_zero_state=True,
148 )
149 self._skip_transpilation = skip_transpilation
150
151 # we need to set the target after the options are set
152 # so that the target_profile can be used to determine
153 # which gates/instructions are available
154 if target is not None:
155 # update the properties so that we are internally consistent
156 self._qiskit_pass_options.update_options(
157 **{
158 "supports_barrier": target.instruction_supported("barrier"),
159 "supports_delay": target.instruction_supported("delay"),
160 "remove_reset_in_zero_state": True,
161 }
162 )
163
164 self._target = target
165 else:
166 self._target = self._build_target()
167
168 self._transpile_options = {}
169
170 basis_gates = None
171 if qasm_export_options is not None and "basis_gates" in qasm_export_options:
172 basis_gates = qasm_export_options.pop("basis_gates")
173 else:
174 # here we get the gates that are in the target but not in qasm's
175 # stdgates so that we can build the basis gates list for the exporter.
176 # A user can override this list by passing in a basis_gates list
177 # We also remove any non-gate instructions from the list.
178 target_gates = set(self.target.operation_names)
179 target_gates -= set(_QISKIT_NON_GATE_INSTRUCTIONS)
180 target_gates -= set(_QISKIT_STDGATES)
181 basis_gates = list(target_gates)
182
183 # set the default options for the exporter
184 self._qasm_export_options = {
185 "includes": ("stdgates.inc",),
186 "alias_classical_registers": False,
187 "allow_aliasing": False,
188 "disable_constants": True,
189 "basis_gates": basis_gates,
190 }
191
192 if qiskit_pass_options is not None:
193 self._qiskit_pass_options.update_options(**qiskit_pass_options)
194 if transpile_options is not None:
195 self._transpile_options.update(**transpile_options)
196 if qasm_export_options is not None:
197 self._qasm_export_options.update(**qasm_export_options)
198
199 def _build_target(self) -> Target:
200 supports_barrier = self._qiskit_pass_options["supports_barrier"]
201 supports_delay = self._qiskit_pass_options["supports_delay"]
202
203 # explicitly set ``num_qubits`` to ``None`` to indicate a :class:`Target` representing a
204 # simulator or other abstract machine that imposes no limits on the number of qubits.
205 return QirTarget.build_target(
206 num_qubits=None,
207 target_profile=self._options["target_profile"],
208 supports_barrier=supports_barrier,
209 supports_delay=supports_delay,
210 )
211
212 @property
213 def target(self) -> Target:
214 """Returns the target of the Backend object."""
215 return self._target
216
217 @property
218 def max_circuits(self):
219 """
220 Returns the maximum number of circuits that can be executed simultaneously.
221 """
222 return None
223
224 @abstractmethod
225 def _execute(self, programs: List[Compilation], **input_params) -> Dict[str, Any]:
226 """Execute circuits on the backend.
227
228 :param programs: Simulator input circuits.
229 :type programs: List[Compilation]
230 :param **input_params: Configuration for simulation/compilation.
231 :return: A dictionary of results.
232 :rtype: dict
233 """
234
235 @abstractmethod
236 def run(
237 self,
238 run_input: Union[QuantumCircuit, List[QuantumCircuit]],
239 **options,
240 ) -> QsJob:
241 pass
242
243 def _run(
244 self,
245 run_input: List[QuantumCircuit],
246 **options,
247 ) -> QsJob:
248 if "name" not in options and len(run_input) == 1:
249 options["name"] = run_input[0].name
250
251 # Get out default options
252 # Look at all of the kwargs and see if they match any of the options
253 # If they do, set the option to the value of the kwarg as an override
254 # We only to remove the options that are in the backend options for
255 # the run so that other options can be passed to other calls.
256 input_params: Dict[str, Any] = vars(self.options).copy()
257 input_params.update(options)
258
259 return self._submit_job(run_input, **input_params)
260
261 def run_job(
262 self, run_input: List[QuantumCircuit], job_id: str, **options
263 ) -> Result:
264 start = monotonic()
265
266 compilations = self._compile(run_input, **options)
267
268 output = self._execute(compilations, **options)
269
270 if not isinstance(output, dict):
271 logger.error("%s: run failed.", self.name)
272 if output:
273 logger.error("Output: %s", output)
274 from .... import QSharpError
275
276 raise QSharpError(str(Errors.RUN_TERMINATED_WITHOUT_OUTPUT))
277
278 output["job_id"] = job_id
279 output["date"] = str(datetime.datetime.now().isoformat())
280 output["status"] = "COMPLETED"
281 output["backend_name"] = self.name
282 output["backend_version"] = self.backend_version
283
284 duration = monotonic() - start
285 output["time_taken"] = str(duration)
286 output["config"] = {
287 "qasm_export_options": str(self._build_qasm_export_options(**options)),
288 "qiskit_pass_options": str(self._build_qiskit_pass_options(**options)),
289 "transpile_options": str(self._build_transpile_options(**options)),
290 }
291 output["header"] = {}
292 return self._create_results(output)
293
294 def _validate_quantum_circuits(
295 self, run_input: Union[QuantumCircuit, List[QuantumCircuit]]
296 ) -> List[QuantumCircuit]:
297 """Normalize and validate run_input to a list of QuantumCircuits.
298
299 Wraps a bare ``QuantumCircuit`` in a list and raises ``ValueError``
300 if any element is not a ``QuantumCircuit``.
301 """
302 if not isinstance(run_input, list):
303 run_input = [run_input]
304 for circuit in run_input:
305 if not isinstance(circuit, QuantumCircuit):
306 raise ValueError(str(Errors.INPUT_MUST_BE_QC))
307 return run_input
308
309 def _submit_job(self, run_input: List[QuantumCircuit], **options) -> QsJob:
310 """Default implementation for simulation backends.
311
312 Submits a ``QsSimJob`` for a single circuit or a ``QsJobSet`` for
313 multiple circuits. Override for backends with different job types
314 (e.g. ``ResourceEstimatorBackend`` uses ``ReJob``).
315 """
316 from uuid import uuid4
317
318 job_id = str(uuid4())
319 executor = options.pop("executor", DetaultExecutor())
320 if len(run_input) == 1:
321 job = QsSimJob(self, job_id, self.run_job, run_input, options, executor)
322 else:
323 job = QsJobSet(self, job_id, self.run_job, run_input, options, executor)
324 job.submit()
325 return job
326
327 def _compile(self, run_input: List[QuantumCircuit], **options) -> List[Compilation]:
328 # for each run input, convert to qasm
329 compilations = []
330 for circuit in run_input:
331 args = options.copy()
332 assert isinstance(
333 circuit, QuantumCircuit
334 ), "Input must be a QuantumCircuit."
335 start = monotonic()
336 qasm = self._qasm(circuit, **args)
337 end = monotonic()
338
339 time_taken = end - start
340 compilation = Compilation(circuit, qasm, time_taken)
341 compilations.append(compilation)
342 return compilations
343
344 def _create_results(self, output: Dict[str, Any]) -> Any:
345 """Default implementation: build a Qiskit ``Result`` from the output dict.
346
347 Override for backends that return a different result type
348 (e.g. ``ResourceEstimatorBackend`` returns ``EstimatorResult``).
349 """
350 return Result.from_dict(output)
351
352 def _map_result_bit(self, v) -> str:
353 """Map a single QIR result value to a bit character.
354
355 Override in subclasses to customize the mapping — for example,
356 to emit a loss marker instead of the default string fallback for
357 unknown values.
358 """
359 from .... import Result as QSharpResult
360
361 if v == QSharpResult.One:
362 return "1"
363 if v == QSharpResult.Zero:
364 return "0"
365 return str(v)
366
367 def _shot_to_bitstring(self, value) -> str:
368 """Recursively convert a QIR shot result to a Qiskit-style bitstring.
369
370 - ``tuple`` → space-joined register parts (multiple classical registers)
371 - ``list`` → concatenated bits via `_map_result_bit`
372 - anything else → ``str(value)``
373 """
374 if isinstance(value, tuple):
375 return " ".join(self._shot_to_bitstring(p) for p in value)
376 elif isinstance(value, list):
377 return "".join(self._map_result_bit(v) for v in value)
378 else:
379 return str(value)
380
381 def _transpile(self, circuit: QuantumCircuit, **options) -> QuantumCircuit:
382 if options.get("skip_transpilation", self._skip_transpilation):
383 return circuit
384
385 circuit = self.run_qiskit_passes(circuit, options)
386
387 transpile_options = self._build_transpile_options(**options)
388 backend = transpile_options.pop("backend", self)
389 target = transpile_options.pop("target", self.target)
390 if get_version_info().startswith("1.2"):
391 # The older Qiskit version does not support the `qubits_initially_zero` option
392 transpiled_circuit = transpile(
393 circuit,
394 backend=backend,
395 target=target,
396 **transpile_options,
397 )
398 else:
399 transpiled_circuit = transpile(
400 circuit,
401 backend=backend,
402 target=target,
403 qubits_initially_zero=True,
404 **transpile_options,
405 )
406 return transpiled_circuit
407
408 def run_qiskit_passes(self, circuit, options):
409 pass_options = self._build_qiskit_pass_options(**options)
410
411 pass_manager = PassManager()
412 if not pass_options["supports_barrier"]:
413 pass_manager.append(RemoveBarriers())
414 if not pass_options["supports_delay"]:
415 pass_manager.append(RemoveDelays())
416 if pass_options["remove_reset_in_zero_state"]:
417 # when doing state initialization, qiskit will reset all qubits to 0
418 # As our semantics are different, we can remove these resets
419 # as it will double the number of qubits if we have to reset them
420 # before using them when using the base profile.
421 pass_manager.append(RemoveResetInZeroState())
422
423 circuit = pass_manager.run(circuit)
424 return circuit
425
426 def _build_qiskit_pass_options(self, **kwargs) -> Dict[str, Any]:
427 params: Dict[str, Any] = vars(self._qiskit_pass_options).copy()
428 for opt in params.copy():
429 if opt in kwargs:
430 params[opt] = kwargs.pop(opt)
431 if "supports_barrier" not in params:
432 params["supports_barrier"] = False
433 if "supports_delay" not in params:
434 params["supports_delay"] = False
435 if "remove_reset_in_zero_state" not in params:
436 params["remove_reset_in_zero_state"] = True
437
438 return params
439
440 def _build_transpile_options(self, **kwargs) -> Dict[str, Any]:
441 # create the default options from the backend
442 args = self._transpile_options.copy()
443 # gather any remaining options that are not in the default list
444 transpile_args = get_transpile_options(**kwargs)
445 args.update(transpile_args)
446 return args
447
448 def _build_qasm_export_options(self, **kwargs) -> Dict[str, Any]:
449 # Disable aliasing until we decide want to support it
450 # The exporter defaults to only having the U gate.
451 # When it sees the stdgates.inc in the default includes list, it adds
452 # bodyless symbols for that fixed gate set.
453 # We set the basis gates for any gates that we want that wouldn't
454 # be defined when stdgates.inc is included.
455
456 # any gates that are not in the stdgates.inc file need to be defined
457 # in the basis gates list passed to the exporter. The exporter doesn't
458 # know about the gates defined in the backend's target.
459 # Anything in the basis_gates gets added to the qasm builder's global
460 # namespace as an opaque gate. All parameter information comes from the
461 # gate object itself in the circuit.
462
463 # create the default options from the backend
464 args = self._qasm_export_options.copy()
465 # gather any remaining options that are not in the default list
466 exporter_args = get_exporter_options(**kwargs)
467 args.update(exporter_args)
468 return args
469
470 def transpile(self, circuit: QuantumCircuit, **options) -> QuantumCircuit:
471 transpiled_circuit = self._transpile(circuit, **options)
472 return transpiled_circuit
473
474 def _qasm(self, circuit: QuantumCircuit, **options) -> str:
475 """Converts a Qiskit QuantumCircuit to QASM 3 for the current backend.
476
477 :param circuit: The QuantumCircuit to be executed.
478 :param **options: Additional options for the transpiler, exporter, or Qiskit passes.
479 Common values include: ``optimization_level``, ``basis_gates``, ``includes``,
480 ``search_path``. Defaults to backend config values.
481 :return: The converted QASM code as a string. Any supplied includes
482 are emitted as ``include`` statements at the top of the program.
483 :rtype: str
484 :raises QasmError: If there is an error generating or parsing QASM.
485 """
486 transpiled_circuit = self.transpile(circuit, **options)
487 try:
488 export_options = self._build_qasm_export_options(**options)
489 exporter = Exporter(**export_options)
490 qasm3_source = exporter.dumps(transpiled_circuit)
491 # Qiskit QASM exporter doesn't handle experimental features correctly and always emits
492 # OPENQASM 3.0; even though switch case is not supported in QASM 3.0, so we bump
493 # the version to 3.1 for now.
494 qasm3_source = qasm3_source.replace("OPENQASM 3.0", "OPENQASM 3.1")
495 return qasm3_source
496 except Exception as ex:
497 from .. import QasmError
498
499 raise QasmError(str(Errors.FAILED_TO_EXPORT_QASM)) from ex
500
501 def _qsharp(self, circuit: QuantumCircuit, **kwargs) -> str:
502 """
503 Converts a Qiskit QuantumCircuit to Q# for the current backend.
504
505 The generated Q# code will not be idiomatic Q# code, but will be
506 a direct translation of the Qiskit circuit.
507
508 :param circuit: The QuantumCircuit to be executed.
509 :param **kwargs: Additional options for the transpiler, exporter, or Qiskit passes.
510 Common values include: ``optimization_level``, ``basis_gates``, ``includes``,
511 ``search_path``, ``output_semantics``. Defaults to backend config values.
512 :return: The converted Q# code as a string.
513 :rtype: str
514 :raises QSharpError: If there is an error evaluating the source code.
515 :raises QasmError: If there is an error generating, parsing, or compiling QASM.
516 """
517
518 qasm_source = self._qasm(circuit, **kwargs)
519
520 args = {
521 "name": kwargs.get("name", circuit.name),
522 }
523
524 if search_path := kwargs.pop("search_path", "."):
525 args["search_path"] = search_path
526
527 if output_semantics := kwargs.pop(
528 "output_semantics", self.options.get("output_semantics", default=None)
529 ):
530 args["output_semantics"] = output_semantics
531
532 qsharp_source = self._qasm_to_qsharp(qasm_source, **args)
533 return qsharp_source
534
535 def qir(
536 self,
537 circuit: QuantumCircuit,
538 **kwargs,
539 ) -> str:
540 """
541 Converts a Qiskit QuantumCircuit to QIR (Quantum Intermediate Representation).
542
543 :param circuit: The input Qiskit QuantumCircuit object.
544 :param **kwargs: Common options:
545
546 - ``target_profile`` (TargetProfile): The target profile for the backend. Defaults to backend config value.
547 - ``output_semantics`` (OutputSemantics): The output semantics for the compilation. Defaults to backend config value.
548 - ``search_path`` (str): The search path for the backend. Defaults to ``'.'``.
549 :return: The converted QIR code as a string.
550 :rtype: str
551 :raises QSharpError: If there is an error evaluating the source code.
552 :raises QasmError: If there is an error generating, parsing, or compiling QASM.
553 :raises ValueError: If the backend configuration does not support QIR generation.
554 """
555 name = kwargs.pop("name", circuit.name)
556 target_profile = kwargs.pop("target_profile", self.options.target_profile)
557 if target_profile == TargetProfile.Unrestricted:
558 raise ValueError(str(Errors.UNRESTRICTED_INVALID_QIR_TARGET))
559
560 qasm_source = self._qasm(circuit, **kwargs)
561
562 args = {
563 "name": name,
564 "target_profile": target_profile,
565 }
566
567 if search_path := kwargs.pop("search_path", "."):
568 args["search_path"] = search_path
569
570 if params := kwargs.pop("params", None):
571 args["params"] = params
572
573 if output_semantics := kwargs.pop(
574 "output_semantics", self.options.get("output_semantics", default=None)
575 ):
576 args["output_semantics"] = output_semantics
577
578 return self._qasm_to_qir(qasm_source, **args)
579
580 def _qasm_to_qir(
581 self,
582 source: str,
583 **kwargs,
584 ) -> str:
585 from ...._native import compile_qasm_program_to_qir
586 from ...._fs import read_file, list_directory, resolve
587 from ...._http import fetch_github
588
589 return compile_qasm_program_to_qir(
590 source,
591 read_file,
592 list_directory,
593 resolve,
594 fetch_github,
595 **kwargs,
596 )
597
598 def _qasm_to_qsharp(
599 self,
600 source: str,
601 **kwargs,
602 ) -> str:
603 from ...._native import compile_qasm_to_qsharp
604 from ...._fs import read_file, list_directory, resolve
605 from ...._http import fetch_github
606
607 return compile_qasm_to_qsharp(
608 source,
609 read_file,
610 list_directory,
611 resolve,
612 fetch_github,
613 **kwargs,
614 )
615