Add CFG plotting and circuit-like export to pyqir.

src/QirTools/pyqir/dev-environment.yml

@@ -0,0 +1,20 @@
+name: pyqir-dev
+channels:
+- conda-forge
+dependencies:
+- maturin
+- tox
+- pip
+- llvm=11
+- llvm-tools=11
+- llvmdev=11
+- notebook
+- matplotlib
+- qutip
+# Qiskit dependencies
+- scipy>=1.0
+- dill
+- scikit-learn
+- pip:
+  - retworkx
+  - qiskit=0.31

src/QirTools/pyqir/pyqir/__init__.py

@@ -2,4 +2,4 @@
 # Licensed under the MIT License.
 
 from .parser import *
-from .builder import *
+from .builder import *

src/QirTools/pyqir/pyqir/export.py

@@ -0,0 +1,230 @@
+# Copyright(c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+# TODO: Support additional simple gates
+# TODO: Support gates w/ classical arguments
+# TODO: Handle nonstandard extensions to OpenQASM better.
+
+from types import MappingProxyType
+from abc import ABCMeta, abstractmethod
+from typing import Any, Callable, Dict, List, Optional, Set, Union, Generic, TypeVar
+from .pyqir import *
+
+try:
+    import qiskit as qk
+except ImportError:
+    qk = None
+
+try:
+    import qutip as qt
+    import qutip.qip.circuit
+    import qutip.qip.operations
+except ImportError:
+    qt = None
+
+TOutput = TypeVar('TOutput')
+
+# TODO: Better names for these utility functions.
+def _resolve_id(value: Any, default: str = "unsupported") -> Union[int, str]:
+    if hasattr(value, "id"):
+        return value.id
+    return default
+
+def _resolve(value, in_):
+    id = _resolve_id(value)
+    if id not in in_:
+        in_[id] = len(in_)
+    return in_[id]
+
+class CircuitLikeExporter(Generic[TOutput], metaclass=ABCMeta):
+    @abstractmethod
+    def on_simple_gate(self, name, *qubits) -> None:
+        pass
+
+    @abstractmethod
+    def on_measure(self, qubit, result) -> None:
+        pass
+
+    @abstractmethod
+    def on_comment(self, text) -> None:
+        pass
+
+    @abstractmethod
+    def export(self) -> TOutput:
+        pass
+
+    @abstractmethod
+    def qubit_as_expr(self, qubit) -> str:
+        pass
+
+    @abstractmethod
+    def result_as_expr(self, result) -> str:
+        pass
+
+class OpenQasm20Exporter(CircuitLikeExporter[str]):
+    header: List[str]
+    lines: List[str]
+
+    qubits: Dict[Any, Optional[int]]
+    results: Dict[Any, Optional[int]]
+
+    qreg = "q"
+    creg = "c"
+
+    def __init__(self, block_name: str):
+        self.lines = []
+        self.header = [
+            f"// Generated from QIR block {block_name}.",
+            "OPENQASM 2.0;",
+            'include "qelib1.inc";',
+            ""
+        ]
+        self.qubits = dict()
+        self.results = dict()
+
+    def qubit_as_expr(self, qubit) -> str:
+        id_qubit = _resolve_id(qubit)
+        if id_qubit not in self.qubits:
+            self.qubits[id_qubit] = len(self.qubits)
+        return f"{self.qreg}[{self.qubits[id_qubit]}]"
+
+    def result_as_expr(self, result) -> str:
+        id_result = _resolve_id(result)
+        if id_result not in self.results:
+            self.results[id_result] = len(self.results)
+        return f"{self.creg}[{self.results[id_result]}]"
+
+    def export(self) -> str:
+        declarations = []
+        if self.qubits:
+            declarations.append(
+                f"qreg {self.qreg}[{len(self.qubits)}];"
+            )
+        if self.results:
+            declarations.append(
+                f"creg {self.creg}[{len(self.results)}];"
+            )
+
+        return "\n".join(
+            self.header + declarations + self.lines
+        )
+
+    def on_comment(self, text) -> None:
+        # TODO: split text into lines first.
+        self.lines.append(f"// {text}")
+
+    def on_simple_gate(self, name, *qubits) -> None:
+        qubit_args = [self.qubit_as_expr(qubit) for qubit in qubits]
+        self.lines.append(f"{name} {', '.join(map(str, qubit_args))};")
+
+    def on_measure(self, qubit, result) -> None:
+        self.lines.append(f"measure {self.qubit_as_expr(qubit)} -> {self.result_as_expr(result)};")
+
+class QiskitExporter(CircuitLikeExporter["qk.QuantumCircuit"]):
+    def __init__(self, block_name: str):
+        self.oq2_exporter = OpenQasm20Exporter(block_name)
+    def qubit_as_expr(self, qubit) -> str:
+        return self.oq2_exporter.qubit_as_expr(qubit)
+    def result_as_expr(self, result) -> str:
+        return self.oq2_exporter.result_as_expr(result)
+    def on_comment(self, text) -> None:
+        return self.oq2_exporter.on_comment(text)
+    def on_measure(self, qubit, result) -> None:
+        return self.oq2_exporter.on_measure(qubit, result)
+    def on_simple_gate(self, name, *qubits) -> None:
+        return self.oq2_exporter.on_simple_gate(name, *qubits)
+    def export(self) -> "qk.QuantumCircuit":
+        return qk.QuantumCircuit.from_qasm_str(self.oq2_exporter.export())
+
+class QuTiPExporter(CircuitLikeExporter["qt.qip.circuit.QubitCircuit"]):
+    actions: List[Callable[["qt.qip.circuit.QubitCircuit"], None]]
+
+    qubits: Dict[Any, int]
+    results: Dict[Any, int]
+
+    gate_names = MappingProxyType({
+        'x': 'X',
+        'y': 'Y',
+        'z': 'Z',
+        'reset': 'reset'
+    })
+
+    def __init__(self):
+        self.actions = []
+        self.qubits = {}
+        self.results = {}
+
+    def export(self) -> TOutput:
+        circuit = qt.qip.circuit.QubitCircuit(N=len(self.qubits), num_cbits=len(self.results))
+        for action in self.actions:
+            action(circuit)
+        return circuit
+
+    def qubit_as_expr(self, qubit) -> str:
+        return repr(_resolve(qubit, self.qubits))
+
+    def result_as_expr(self, result) -> str:
+        return repr(_resolve(result, self.results))
+
+    def on_simple_gate(self, name, *qubits) -> None:
+        targets = [_resolve(qubit, self.qubits) for qubit in qubits]
+        self.actions.append(lambda circuit:
+            circuit.add_gate(
+                self.gate_names[name],
+                targets=targets
+            )
+        )
+
+    def on_measure(self, qubit, result) -> None:
+        targets = [_resolve(qubit, self.qubits)]
+        classical_store = _resolve(result, self.results)
+        self.actions.append(lambda circuit:
+            circuit.add_measurement(
+                "MZ",
+                targets=targets,
+                classical_store=classical_store
+            )
+        )
+
+    def on_comment(self, text) -> None:
+        print(f"# {text}")
+
+def export_to(block, exporter: CircuitLikeExporter[TOutput]) -> Optional[TOutput]:
+    if not block.is_circuit_like:
+        return None
+
+    qis_gate_mappings = {
+        '__quantum__qis__x__body': 'x',
+        '__quantum__qis__y__body': 'y',
+        '__quantum__qis__z__body': 'z',
+        '__quantum__qis__h__body': 'h',
+        '__quantum__qis__cnot__body': 'cnot'
+    }
+
+    for instruction in block.instructions:
+        # Translate simple gates (that is, only quantum args).
+        if instruction.func_name in qis_gate_mappings:
+            exporter.on_simple_gate(qis_gate_mappings[instruction.func_name], *instruction.func_args)
+
+        # Reset requires nonstandard support.
+        elif instruction.func_name == "__quantum__qis__reset__body":
+            exporter.on_comment("Requires nonstandard reset gate:")
+            exporter.on_simple_gate("reset", *instruction.func_args)
+
+        # Measurements are special in OpenQASM 2.0, so handle them here.
+        elif instruction.func_name == "__quantum__qis__mz__body":
+            target, result = instruction.func_args
+            exporter.on_measure(target, result)
+            # target = lookup_qubit(target)
+            # result = lookup_result(result)
+            # lines.append(f"measure {target} -> {result};")
+
+        # Handle special cases of QIR functions as needed.
+        elif instruction.func_name == "__quantum__qir__read_result":
+            exporter.on_comment(f"%{instruction.output_name} = {exporter.result_as_expr(instruction.func_args[0])}")
+
+        else:
+            exporter.on_comment("// Unsupported QIS operation:")
+            exporter.on_comment(f"// {instruction.func_name} {', '.join(map(str, instruction.func_args))}")
+
+    return exporter.export()

src/QirTools/pyqir/pyqir/parser.py

@@ -2,8 +2,30 @@
 # Licensed under the MIT License.
 
 from .pyqir import *
+from . import export
 from typing import List, Optional, Tuple
 
+try:
+    import retworkx as rx
+except ImportError:
+    rx = None
+
+try:
+    import retworkx.visualization as rxv
+except ImportError:
+    rxv = None
+
+try:
+    import qiskit as qk
+except ImportError:
+    qk = None
+
+try:
+    import qutip as qt
+    import qutip.qip.circuit
+    import qutip.qip.operations
+except ImportError:
+    qt = None
 
 class QirType:
     """
@@ -748,6 +770,30 @@ def get_phi_pairs_by_source_name(self, name: str) -> List[Tuple[str, QirOperand]
         """
         return [(p[0], QirOperand(p[1])) for p in self.block.get_phi_pairs_by_source_name(name)]
 
+    @property
+    def is_circuit_like(self) -> bool:
+        return all(
+            isinstance(instruction, (QirQisCallInstr, QirQirCallInstr))
+            for instruction in self.instructions
+        )
+        # TODO: Check all args are qubit constants.
+
+    def as_openqasm_20(self) -> Optional[str]:
+        """
+        If this block is circuit-like (that is, consists only of quantum instructions),
+        converts it to a representation in OpenQASM 2.0; otherwise, returns `None`.
+
+        Note that the returned representation does not include leading phi nodes, nor trailing terminators.
+        """
+        return export.export_to(self, export.OpenQasm20Exporter(self.name))
+
+    def as_qiskit_circuit(self) -> Optional["qk.QuantumCircuit"]:
+        return export.export_to(self, export.QiskitExporter(self.name))
+
+    def as_qutip_circuit(self) -> Optional["qt.qip.circuit.QubitCircuit"]:
+        return export.export_to(self, export.QuTiPExporter())
+
+
 class QirParameter:
     """
     Instances of the QirParameter type describe a parameter in a function definition or declaration. They
@@ -781,6 +827,9 @@ class QirFunction:
     def __init__(self, func: PyQirFunction):
         self.func = func
 
+    def __repr__(self) -> str:
+        return f"<QIR function {self.name} at {id(self):0x}>"
+
     @property
     def name(self) -> str:
         """
@@ -856,6 +905,37 @@ def get_instruction_by_output_name(self, name: str) -> Optional[QirInstr]:
             return QirInstr(instr)
         return None
 
+    def control_flow_graph(self) -> "rx.Digraph":
+        cfg = rx.PyDiGraph(check_cycle=False, multigraph=True)
+        blocks = self.blocks
+        block_indices = {
+            block.name: cfg.add_node(block)
+            for block in blocks
+        }
+
+        idx_return = cfg.add_node("Return")
+        idx_bottom = None
+
+        for idx_block, block in enumerate(blocks):
+            term = block.terminator
+            if isinstance(term, QirCondBrTerminator):
+                cfg.add_edge(idx_block, block_indices[term.true_dest], True)
+                cfg.add_edge(idx_block, block_indices[term.false_dest], False)
+            elif isinstance(term, QirBrTerminator):
+                cfg.add_edge(idx_block, block_indices[term.dest], ())
+            elif isinstance(term, QirRetTerminator):
+                cfg.add_edge(idx_block, idx_return, ())
+            elif isinstance(term, QirSwitchTerminator):
+                print(f"Not yet implemented: {term}")
+            elif isinstance(term, QirUnreachableTerminator):
+                if idx_bottom is None:
+                    idx_bottom = cfg.add_node("⊥")
+                cfg.add_edge(idx_block, idx_bottom)
+            else:
+                print(f"Not yet implemented: {term}")
+
+        return cfg
+
 class QirModule:
     """
     Instances of QirModule parse a QIR program from bitcode into an in-memory
@@ -910,4 +990,3 @@ def interop_funcs(self) -> List[QirFunction]:
         Gets any functions with the "InteropFriendly" attribute.
         """
         return [QirFunction(i) for i in self.module.get_interop_funcs()]
-