QASM Import

quanta.export.qasm_import

quanta.export.qasm_import -- QASM 2.0/3.0 to Quanta DAG import pipeline.

Parses OpenQASM files and converts them to Quanta's internal DAG representation. Supports OpenQASM 3.0 dynamic circuits including mid-circuit measurement and classical feedforward condition execution.

Supports

• QASM 2.0 (qelib1.inc gates) and QASM 3.0 (stdgates.inc)
• Standard gates: h, x, y, z, s, t, cx, cz, swap, ccx, rx, ry, rz
• Qubit and bit register declarations (qubit[N], bit[N], qreg, creg)
• Parametric gates with float params
• Mid-circuit measurements: c[i] = measure q[j]; and measure q[j] -> c[i];
• Conditional gate execution: if (c[i] == 1) { gate q; } and if (c == 1) gate q;

Example

from quanta.export.qasm_import import from_qasm dag = from_qasm(''' ... OPENQASM 3.0; ... include "stdgates.inc"; ... qubit[2] q; ... bit[1] c; ... h q[0]; ... c[0] = measure q[0]; ... if (c[0] == 1) { x q[1]; } ... ''') print(dag.gate_count()) 3

ConditionalOpNode

Bases: OpNode

OpNode extended with condition and cbit fields for dynamic circuits.

Source code in quanta/export/qasm_import.py

class ConditionalOpNode(OpNode):
    """OpNode extended with condition and cbit fields for dynamic circuits."""
    __slots__ = ("condition", "cbit")

    def __init__(
        self,
        node_id: int,
        gate_name: str,
        qubits: tuple[int, ...],
        params: tuple[float, ...] = (),
        condition: tuple[int, int] | None = None,
        cbit: int | None = None,
    ) -> None:
        super().__init__(node_id=node_id, gate_name=gate_name, qubits=qubits, params=params)
        object.__setattr__(self, "condition", condition)
        object.__setattr__(self, "cbit", cbit)

DynamicDAGCircuit

Bases: DAGCircuit

DAGCircuit supporting dynamic circuits with classical conditions and registers.

Source code in quanta/export/qasm_import.py

class DynamicDAGCircuit(DAGCircuit):
    """DAGCircuit supporting dynamic circuits with classical conditions and registers."""
    __slots__ = ("num_clbits", "conditions", "instructions")

    def __init__(self, num_qubits: int, num_clbits: int = 0) -> None:
        super().__init__(num_qubits)
        self.num_clbits = num_clbits
        self.conditions: dict[int, tuple[int, int]] = {}
        self.instructions: list[Instruction] = []

from_qasm

from_qasm(qasm_str: str) -> DAGCircuit

Parses QASM 2.0/3.0 string and returns a Quanta DAGCircuit (or DynamicDAGCircuit).

Parameters:

Name	Type	Description	Default
qasm_str	str	OpenQASM string (2.0 or 3.0).	required

Returns:

Type	Description
DAGCircuit	DAGCircuit ready for simulation or compilation.

Source code in quanta/export/qasm_import.py

def from_qasm(qasm_str: str) -> DAGCircuit:
    """Parses QASM 2.0/3.0 string and returns a Quanta DAGCircuit (or DynamicDAGCircuit).

    Args:
        qasm_str: OpenQASM string (2.0 or 3.0).

    Returns:
        DAGCircuit ready for simulation or compilation.
    """
    statements = _split_qasm_statements(qasm_str)

    num_qubits = 0
    num_cbits = 0
    raw_ops: list[tuple] = []
    active_condition: tuple[int, int] | None = None
    has_final_measure = False
    qubit_offset: dict[str, int] = {}
    cbit_offset: dict[str, int] = {}

    for stmt in statements:
        if stmt == "}":
            active_condition = None
            continue

        clean = stmt.rstrip(";").strip()
        if not clean:
            continue

        # Skip headers
        if clean.startswith(("OPENQASM", "include")):
            continue

        # QASM 2.0: qreg q[N];
        m_q2 = re.match(r"qreg\s+(\w+)\s*\[\s*(\d+)\s*\]", clean)
        if m_q2:
            reg_name, size = m_q2.group(1), int(m_q2.group(2))
            qubit_offset[reg_name] = num_qubits
            num_qubits += size
            continue

        # QASM 3.0: qubit[N] q;
        m_q3 = re.match(r"qubit\s*\[\s*(\d+)\s*\]\s+(\w+)", clean)
        if m_q3:
            size, reg_name = int(m_q3.group(1)), m_q3.group(2)
            qubit_offset[reg_name] = num_qubits
            num_qubits += size
            continue

        # Classical registers: creg c[N]; or bit[N] c; or bit c;
        m_c = re.match(r"(?:creg|bit)\s*(?:\[\s*(\d+)\s*\])?\s*(\w+)", clean)
        if m_c and (clean.startswith("creg") or clean.startswith("bit")):
            size_str, reg_name = m_c.group(1), m_c.group(2)
            size = int(size_str) if size_str is not None else 1
            cbit_offset[reg_name] = num_cbits
            num_cbits += size
            continue

        # Barrier (skip)
        if clean.startswith("barrier"):
            continue

        # Conditional branch: if (c[i] == val) { ... } or if (c == val) ...
        m_if = re.match(
            r"if\s*\(\s*(\w+)(?:\[(\d+)\])?\s*==\s*(\d+)\s*\)\s*(\{?)\s*(.*)",
            clean,
        )
        if m_if:
            reg, idx_str, val_str, has_brace, raw_body = m_if.groups()
            c_idx = cbit_offset.get(reg, 0) + (int(idx_str) if idx_str is not None else 0)
            cond = (c_idx, int(val_str))
            has_closing = "}" in raw_body
            active_condition = cond if (has_brace and not has_closing) else None

            body = raw_body.rstrip("};").strip()
            if body:
                pm = _parse_measure_line(body, qubit_offset, cbit_offset)
                if pm:
                    raw_ops.append(("measure", pm[0], pm[1], cond))
                else:
                    pg = _parse_gate_line(body, qubit_offset)
                    if pg:
                        raw_ops.append((
                            "gate",
                            Instruction(
                                gate_name=pg.gate_name,
                                qubits=pg.qubits,
                                params=pg.params,
                                condition=cond,
                            ),
                        ))
            continue

        # If inside a multi-line conditional block
        if active_condition is not None:
            pm = _parse_measure_line(clean, qubit_offset, cbit_offset)
            if pm:
                raw_ops.append(("measure", pm[0], pm[1], active_condition))
            else:
                pg = _parse_gate_line(clean, qubit_offset)
                if pg:
                    raw_ops.append((
                        "gate",
                        Instruction(
                            gate_name=pg.gate_name,
                            qubits=pg.qubits,
                            params=pg.params,
                            condition=active_condition,
                        ),
                    ))
            continue

        # Mid-circuit measurement targeting specific classical bit
        pm = _parse_measure_line(clean, qubit_offset, cbit_offset)
        if pm:
            raw_ops.append(("measure", pm[0], pm[1], None))
            continue

        # Full register measurement (e.g., measure q -> c; or c = measure q;)
        if "measure" in clean:
            raw_ops.append(("final_measure", None))
            continue

        # Regular gate line
        pg = _parse_gate_line(clean, qubit_offset)
        if pg:
            raw_ops.append((
                "gate",
                Instruction(
                    gate_name=pg.gate_name,
                    qubits=pg.qubits,
                    params=pg.params,
                    condition=None,
                ),
            ))

    has_dynamic_features = any(
        (op[0] == "measure" and any(raw_ops[j][0] == "gate" for j in range(idx + 1, len(raw_ops))))
        or (len(op) > 3 and op[3] is not None)
        for idx, op in enumerate(raw_ops)
    )

    # Second pass: classify measurements as mid-circuit vs terminal
    instructions: list[Instruction] = []
    for _i, op in enumerate(raw_ops):
        op_type = op[0]
        if op_type == "gate":
            instructions.append(op[1])
        elif op_type == "final_measure":
            has_final_measure = True
        elif op_type == "measure":
            qubits, cbit, cond = op[1], op[2], op[3]
            if has_dynamic_features:
                instructions.append(
                    Instruction(gate_name="measure", qubits=qubits, condition=cond, cbit=cbit)
                )
            else:
                has_final_measure = True

    if num_qubits == 0:
        num_qubits = _infer_qubits(instructions)

    # Check if this circuit uses dynamic features (conditions or mid-circuit measurements)
    is_dynamic = any(inst.condition is not None or inst.cbit is not None for inst in instructions)

    if not is_dynamic:
        # Static circuit: use standard CircuitBuilder & DAGCircuit for 100% backward compatibility
        builder = CircuitBuilder(num_qubits)
        for inst in instructions:
            builder.record(inst)
        if has_final_measure:
            builder.measurement = MeasureSpec(qubits=tuple(range(num_qubits)))
        return DAGCircuit.from_builder(builder)

    # Dynamic circuit: build DynamicDAGCircuit with causal dependency tracking
    dag = DynamicDAGCircuit(num_qubits, num_clbits=num_cbits)
    dag.instructions = instructions

    last_on_qubit: dict[int, int] = {
        q: dag._input_nodes[q].node_id
        for q in range(num_qubits)
    }
    last_on_cbit: dict[int, int] = {}

    for inst in instructions:
        node_id = dag._node_counter
        op = ConditionalOpNode(
            node_id=node_id,
            gate_name=inst.gate_name,
            qubits=inst.qubits,
            params=inst.params,
            condition=inst.condition,
            cbit=inst.cbit,
        )
        dag._nodes[node_id] = op
        dag._node_counter += 1

        if inst.condition is not None:
            dag.conditions[node_id] = inst.condition
            cbit_idx = inst.condition[0]
            if cbit_idx in last_on_cbit:
                dag._add_edge(last_on_cbit[cbit_idx], node_id)

        if inst.cbit is not None:
            last_on_cbit[inst.cbit] = node_id

        for q in inst.qubits:
            if q in last_on_qubit:
                dag._add_edge(last_on_qubit[q], node_id)
                last_on_qubit[q] = node_id

    for q in range(num_qubits):
        out = dag._add_node(OutputNode(node_id=0, qubit=q))
        dag._output_nodes[q] = out
        dag._add_edge(last_on_qubit[q], out.node_id)

    if has_final_measure:
        dag.measurement = MeasureSpec(qubits=tuple(range(num_qubits)))

    return dag

from_qasm_file

from_qasm_file(filepath: str) -> DAGCircuit

Loads a QASM file and returns a DAGCircuit.

Source code in quanta/export/qasm_import.py

def from_qasm_file(filepath: str) -> DAGCircuit:
    """Loads a QASM file and returns a DAGCircuit."""
    with open(filepath) as f:
        return from_qasm(f.read())