Repository: PennyLaneAI/pennylane
Pull request #7197 (2025-04-03 19:47:09)

Add ToBloq adapter class and to_bloq function to convert PennyLane circuits and operators to Qualtran bloqs
**Context:**
We implement the adapter class `ToBloq` and a user-facing function `to_bloq` that converts a PennyLane Operator or QNode to a Qualtran bloq/cbloq. 

In this PR, we specifically target the ability to convert PennyLane decompositions to their Qualtran equivalent. When a PennyLane Operator is wrapped with the ToBloq class or is wrapped with `qml.to_bloq(..., map_ops=False)`, `decompose_bloq()` will return the PennyLane decomposition. This enables us to use functions in Qualtran, such as `draw_musical_score`, `call_graph()` and many more. 

Important to note that calling `to_bloq()` on basic gates that have **direct** equivalents (e.g. Hadamard, RX...) will **always** return the Qualtran equivalent, even if `map_ops=False`. We do this because in order for some Qualtran functionality to work properly, it needs to see exactly `qt.Hadamard()` rather than `qml.ToBloq(qml.Hadamard())`; if for some reason, someone really wants to use the wrapped bloq, they can always call `qml.ToBloq()` on the operator directly to "force" the wrapping.

We also implement the outlines for more bespoke functionality, namely implementing `_get_op_call_graph()` for QPE and expanding `_map_to_bloq()` to have a default and custom mapping options. We further expand on these functionalities in the wrapper PR (#7536) and the mapper PR (#7604).

**Description of the Change:**
Implementation of the `ToBloq` class, `to_bloq()` function, and `_map_to_bloq()` function. Also includes stubs for `_get_op_call_graph()`. There is also a number of tests for all the functions, including those that verify that the "round-trip" works, and that the decompositions are accurate. 

**Benefits:**
PL ops and qnodes can be translated to Bloqs, which can then be used as-is for deeper analysis in Qualtran.

**Related GitHub Issues:**
[[sc-87435](https://app.shortcut.com/xanaduai/story/87435)]
[sc-89901]

austingmhuang: [sc-87435]
JerryChen97: > [[sc-87435](https://app.shortcut.com/xanaduai/story/87435)]

Mind move this to the bottom of PR description? In this way devs in the future can immediately see the attached SC link using gitblame
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/PennyLaneAI/pennylane/pull/7197?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=PennyLaneAI) Report
Attention: Patch coverage is `8.10811%` with `102 lines` in your changes missing coverage. Please review.
> Project coverage is 99.44%. Comparing base [(`60421c7`)](https://app.codecov.io/gh/PennyLaneAI/pennylane/commit/60421c750d6fdd3b3c662990ee538aae5acc82e2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=PennyLaneAI) to head [(`2d45445`)](https://app.codecov.io/gh/PennyLaneAI/pennylane/commit/2d45445eb58321b36e2e93e72a8e89ac1c22a376?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=PennyLaneAI).

| [Files with missing lines](https://app.codecov.io/gh/PennyLaneAI/pennylane/pull/7197?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=PennyLaneAI) | Patch % | Lines |
|---|---|---|
| [pennylane/io/qualtran\_io.py](https://app.codecov.io/gh/PennyLaneAI/pennylane/pull/7197?src=pr&el=tree&filepath=pennylane%2Fio%2Fqualtran_io.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=PennyLaneAI#diff-cGVubnlsYW5lL2lvL3F1YWx0cmFuX2lvLnB5) | 7.27% | [102 Missing :warning: ](https://app.codecov.io/gh/PennyLaneAI/pennylane/pull/7197?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=PennyLaneAI) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #7197      +/-   ##
==========================================
- Coverage   99.64%   99.44%   -0.21%     
==========================================
  Files         513      513              
  Lines       50182    50290     +108     
==========================================
+ Hits        50006    50012       +6     
- Misses        176      278     +102     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/PennyLaneAI/pennylane/pull/7197?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=PennyLaneAI).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=PennyLaneAI).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>

Patch:
diff --git a/doc/code/qml_io.rst b/doc/code/qml_io.rst
index b5e7ab4b1..205560186 100644
--- a/doc/code/qml_io.rst
+++ b/doc/code/qml_io.rst
@@ -23,6 +23,7 @@ Functions
     ~from_qiskit_op
     ~from_quil
     ~from_quil_file
+    ~to_bloq
 
 Classes
 ^^^^^^^
@@ -30,4 +31,5 @@ Classes
 .. autosummary::
     :toctree: api
 
-    ~FromBloq
\ No newline at end of file
+    ~FromBloq
+    ~ToBloq
diff --git a/doc/introduction/dynamic_quantum_circuits.rst b/doc/introduction/dynamic_quantum_circuits.rst
index 12398eb2e..5f69c541b 100644
--- a/doc/introduction/dynamic_quantum_circuits.rst
+++ b/doc/introduction/dynamic_quantum_circuits.rst
@@ -266,7 +266,7 @@ scalings  with respect to the number of mid-circuit measurements (and shots) are
 +--------------------------+-------------------------------------------+-----------------------------------------------------------+-------------------------------------------+--------------+--------------+
 | Dynamic one-shot         | :gr:`\ ` :math:`\mathcal{O}(1)`           | :rd:`\ ` :math:`\mathcal{O}(n_{shots})`                   | :or:`\ ` finite differences\ :math:`{}^2` | :gr:`\ ` yes | :rd:`\ ` no  |
 +--------------------------+-------------------------------------------+-----------------------------------------------------------+-------------------------------------------+--------------+--------------+
-| Tree-traversal           | :or:`\ ` :math:`\mathcal{O}(n_{MCM}+1)`   | :or:`\ ` :math:`\mathcal{O}(min(n_{shots}, 2^{n_{MCM}}))` | :or:`\ ` finite differences\ :math:`{}^2` | :gr:`\ ` yes | :rd:`\ ` no  |
+| Tree-traversal           | :or:`\ ` :math:`\mathcal{O}(n_{MCM}+1)`   | :or:`\ ` :math:`\mathcal{O}(min(n_{shots}, 2^{n_{MCM}}))` | :or:`\ ` finite differences\ :math:`{}^2` | :gr:`\ ` yes | :gr:`\ ` yes |
 +--------------------------+-------------------------------------------+-----------------------------------------------------------+-------------------------------------------+--------------+--------------+
 
 
@@ -393,11 +393,19 @@ Since the counts of many nodes come out to be zero for shot-based simulations,
 it is often possible to ignore entire sub-trees, thereby reducing the computational
 cost.
 
-.. warning::
+.. note::
 
-    The tree-traversal algorithm is only supported by the
-    :class:`~.pennylane.devices.DefaultQubit` device, and currently does
-    not support just-in-time (JIT) compilation.
+    The tree-traversal algorithm is supported by the following devices:
+
+    * :class:`~.pennylane.devices.DefaultQubit`,
+    
+    * `lightning.qubit <https://docs.pennylane.ai/projects/lightning/en/stable/lightning_qubit/device.html>`_,
+    
+    * `lightning.gpu <https://docs.pennylane.ai/projects/lightning/en/stable/lightning_gpu/device.html>`_,
+    
+    * `lightning.kokkos <https://docs.pennylane.ai/projects/lightning/en/stable/lightning_kokkos/device.html>`_,
+    
+    and currently does not support just-in-time (JIT) compilation.
 
 .. _mcm_config:
 
diff --git a/doc/releases/changelog-dev.md b/doc/releases/changelog-dev.md
index 97dbc28be..057e27e56 100644
--- a/doc/releases/changelog-dev.md
+++ b/doc/releases/changelog-dev.md
@@ -185,6 +185,71 @@
   [(#7355)](https://github.com/PennyLaneAI/pennylane/pull/7355)
   [(#7586)](https://github.com/PennyLaneAI/pennylane/pull/7586)
 
+<h4>Qualtran Integration 🔗</h4>
+
+* It's now possible to convert PennyLane operators to [Qualtran](https://qualtran.readthedocs.io/en/latest/) bloqs with the new :func:`qml.to_bloq <pennylane.to_bloq>` function. 
+  [(#7197)](https://github.com/PennyLaneAI/pennylane/pull/7197)
+  [(#7604)](https://github.com/PennyLaneAI/pennylane/pull/7604)
+  [(#7536)](https://github.com/PennyLaneAI/pennylane/pull/7536)
+  :func:`qml.to_bloq <pennylane.to_bloq>` translates PennyLane operators into equivalent [Qualtran bloqs](https://qualtran.readthedocs.io/en/latest/bloqs/index.html#bloqs-library). It requires one input and takes in two optional inputs:
+  * circuit (QNode| Qfunc | Operation): a PennyLane ``QNode``, ``Qfunc``, or operator to be wrapped as a Qualtran Bloq.
+  * map_ops (bool): Whether to map operations to a Qualtran Bloq. Operations are wrapped as a ``ToBloq`` when False. Default is True.
+  * custom_mapping (dict): Dictionary to specify a mapping between a PennyLane operator and a Qualtran Bloq. A default mapping is used if not defined.
+  The following example converts a PennyLane Operator into a Qualtran Bloq:
+
+  ```python
+  import pennylane as qml
+  from qualtran.drawing import get_musical_score_data, draw_musical_score, show_bloq
+
+  control_wires = [2, 3]
+  estimation_wires = [4, 5, 6, 7, 8, 9]
+
+  H = -0.4 * qml.Z(0) + 0.3 * qml.Z(1) + 0.4 * qml.Z(0) @ qml.Z(1)
+
+  op = qml.QuantumPhaseEstimation(
+      qml.Qubitization(H, control_wires), estimation_wires=estimation_wires
+  )
+
+  cbloq = qml.to_bloq(op).decompose_bloq()
+  fig, ax = draw_musical_score(get_musical_score_data(cbloq))
+  show_bloq(cbloq)
+  ```
+
+  Let's define a custom mapping instead.
+
+  ```python
+  from qualtran.bloqs.phase_estimation import LPResourceState
+  from qualtran.bloqs.phase_estimation.text_book_qpe import TextbookQPE
+
+  custom_map = {
+    op: TextbookQPE(
+        unitary=qml.to_bloq(qml.Qubitization(H, control_wires)), 
+        ctrl_state_prep=LPResourceState(len(estimation_wires))
+    )
+  }
+
+  cbloq = qml.to_bloq(op, map_ops=True, custom_mapping=custom_map).decompose_bloq()
+  draw_musical_score(get_musical_score_data(cbloq))
+  show_bloq(cbloq)
+  ```
+
+  Alternatively, rather than map directly to a Qualtran Bloq, we can preserve the original
+  PennyLane decomposition by setting `map_ops` to False.
+
+  ```python
+  op_wrapped_as_bloq = qml.to_bloq(op, map_ops=False)
+  cbloq = op_wrapped_as_bloq.decompose_bloq()
+  draw_musical_score(get_musical_score_data(cbloq))
+  show_bloq(cbloq)
+
+  # We can also leverage Qualtran features to get resource counts and call graphs, among other things
+  from qualtran.drawing import show_call_graph, show_counts_sigma  
+
+  graph, sigma = qml.to_bloq(op, map_ops=True).call_graph()
+  show_call_graph(graph)
+  show_counts_sigma(sigma)
+  ```
+
 * A new template :class:`~.SemiAdder` has been added, allowing for quantum-quantum in-place addition.
   This operator performs the plain addition of two integers in the computational basis.
   [(#7494)](https://github.com/PennyLaneAI/pennylane/pull/7494)
@@ -991,6 +1056,11 @@ Here's a list of deprecations made this release. For a more detailed breakdown o
   off from the original tape by a global phase.
   [(#7619)](https://github.com/PennyLaneAI/pennylane/pull/7619)
 
+* Updated documentation for mid-circuit measurements using the Tree Traversal algorithm
+  to reflect supported devices and usage in analytic simulations,
+  in the :doc:`/introduction/dynamic_quantum_circuits` page.
+  [(#7691)](https://github.com/PennyLaneAI/pennylane/pull/7691)
+
 <h3>Contributors ✍️</h3>
 
 This release contains contributions from (in alphabetical order):
@@ -1005,7 +1075,9 @@ Pietropaolo Frisoni,
 Simone Gasperini,
 Korbinian Kottmann,
 Christina Lee,
+Austin Huang,
 Anton Naim Ibrahim,
+Luis Alfredo Nuñez Meneses
 Oumarou Oumarou,
 Lee J. O'Riordan,
 Mudit Pandey,
diff --git a/pennylane/__init__.py b/pennylane/__init__.py
index addc1749b..b37ecd019 100644
--- a/pennylane/__init__.py
+++ b/pennylane/__init__.py
@@ -76,19 +76,6 @@ from pennylane.about import about
 from pennylane.circuit_graph import CircuitGraph
 from pennylane.configuration import Configuration
 from pennylane.registers import registers
-from pennylane.io import (
-    from_pyquil,
-    from_qasm,
-    to_openqasm,
-    from_qiskit,
-    from_qiskit_noise,
-    from_qiskit_op,
-    from_quil,
-    from_quil_file,
-    FromBloq,
-    bloq_registers,
-    from_qasm3,
-)
 from pennylane.measurements import (
     counts,
     density_matrix,
@@ -116,6 +103,21 @@ from pennylane.templates.state_preparations import *
 from pennylane.templates.subroutines import *
 from pennylane import qaoa
 from pennylane.workflow import QNode, qnode, execute, set_shots
+from pennylane.io import (
+    from_pyquil,
+    from_qasm,
+    to_openqasm,
+    from_qiskit,
+    from_qiskit_noise,
+    from_qiskit_op,
+    from_quil,
+    from_quil_file,
+    FromBloq,
+    bloq_registers,
+    from_qasm3,
+    to_bloq,
+    ToBloq,
+)
 from pennylane.transforms import (
     transform,
     batch_params,
diff --git a/pennylane/io/__init__.py b/pennylane/io/__init__.py
index 3780da2b3..e57b103f1 100644
--- a/pennylane/io/__init__.py
+++ b/pennylane/io/__init__.py
@@ -27,4 +27,4 @@ from .io import (
     plugin_converters,
     from_qasm3,
 )
-from .qualtran_io import FromBloq, bloq_registers
+from .qualtran_io import FromBloq, bloq_registers, to_bloq, ToBloq
diff --git a/pennylane/io/qualtran_io.py b/pennylane/io/qualtran_io.py
index bf569c75f..1dcd431e3 100644
--- a/pennylane/io/qualtran_io.py
+++ b/pennylane/io/qualtran_io.py
@@ -19,22 +19,661 @@ This submodule contains the adapter class for Qualtran-PennyLane interoperabilit
 # pylint: disable=unused-argument
 
 from collections import defaultdict
-from functools import lru_cache, singledispatch
+from functools import cached_property, singledispatch, wraps
+from typing import Dict, List, Tuple
 
 import numpy as np
 
+import pennylane.measurements as qmeas
 import pennylane.ops as qops
-from pennylane.operation import DecompositionUndefinedError, MatrixUndefinedError, Operation
+import pennylane.templates as qtemps
+from pennylane import math
+from pennylane.operation import (
+    DecompositionUndefinedError,
+    MatrixUndefinedError,
+    Operation,
+    Operator,
+)
+from pennylane.queuing import AnnotatedQueue, QueuingManager
 from pennylane.registers import registers
+from pennylane.tape import make_qscript
+from pennylane.templates.state_preparations.superposition import _assign_states
 from pennylane.wires import WiresLike
+from pennylane.workflow import construct_tape
+from pennylane.workflow.qnode import QNode
 
 try:
+    import cirq
     import qualtran as qt
+    from qualtran import Bloq, CtrlSpec
+    from qualtran._infra.gate_with_registers import split_qubits
+    from qualtran.bloqs import basic_gates as qt_gates
+
+    qualtran = True
 except (ModuleNotFoundError, ImportError) as import_error:
-    pass
+    qualtran = False
+
+    Bloq = object
+
+
+@singledispatch
+def _get_op_call_graph(op):
+    """Return call graph for PennyLane Operator. If the call graph is not implemented,
+    return ``None``, which means we will build the call graph via decomposition"""
+
+    # TODO: Integrate with resource operators and the new decomposition pipelines
+    return None
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.subroutines.qpe.QuantumPhaseEstimation):
+    """Call graph for Quantum Phase Estimation"""
+
+    # From ResourceQFT
+    gate_counts = defaultdict(int, {})
+
+    gate_counts[qt_gates.Hadamard()] = len(op.estimation_wires)
+    controlled_unitary = _map_to_bloq(op.hyperparameters["unitary"]).controlled(CtrlSpec(cvs=[1]))
+    gate_counts[controlled_unitary] = (2 ** len(op.estimation_wires)) - 1
+    adjoint_qft = _map_to_bloq(qtemps.QFT(wires=op.estimation_wires), map_ops=False).adjoint()
+    gate_counts[adjoint_qft] = 1
+
+    return gate_counts
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.subroutines.TrotterizedQfunc):
+    """Call graph for qml.trotterize"""
+
+    # From ResourceTrotterizedQfunc
+    n = op.hyperparameters["n"]
+    order = op.hyperparameters["order"]
+    k = order // 2
+    qfunc = op.hyperparameters["qfunc"]
+    qfunc_args = op.parameters[1:]
+    base_hyper_params = ("n", "order", "qfunc", "reverse")
+
+    with QueuingManager.stop_recording():
+        with AnnotatedQueue() as q:
+            qfunc_args = op.parameters
+            qfunc_kwargs = {
+                k: v for k, v in op.hyperparameters.items() if not k in base_hyper_params
+            }
+
+            qfunc = op.hyperparameters["qfunc"]
+            qfunc(*qfunc_args, wires=op.wires, **qfunc_kwargs)
+
+    call_graph = defaultdict(int, {})
+    if order == 1:
+        for q_op in q.queue:
+            call_graph[_map_to_bloq(q_op)] += 1
+        return call_graph
+
+    num_gates = 2 * n * (5 ** (k - 1))
+    for q_op in q.queue:
+        call_graph[_map_to_bloq(q_op)] += num_gates
+
+    return call_graph
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.state_preparations.Superposition):
+    """Call graph for Superposition"""
+
+    # From ResourceSuperposition
+    gate_types = defaultdict(int, {})
+    wires = op.wires
+    coeffs = op.coeffs
+    bases = op.hyperparameters["bases"]
+    num_basis_states = len(bases)
+    size_basis_state = len(bases[0])  # assuming they are all the same size
+
+    dic_state = dict(zip(bases, coeffs))
+    perms = _assign_states(bases)
+    new_dic_state = {perms[key]: val for key, val in dic_state.items() if key in perms}
+
+    sorted_coefficients = [
+        value
+        for _, value in sorted(
+            new_dic_state.items(), key=lambda item: int("".join(map(str, item[0])), 2)
+        )
+    ]
+    msp = qops.StatePrep(
+        math.stack(sorted_coefficients),
+        wires=wires[-int(math.ceil(math.log2(len(coeffs)))) :],
+        pad_with=0,
+    )
+    gate_types[_map_to_bloq(msp)] = 1
+
+    cnot = qt_gates.CNOT()
+    num_zero_ctrls = size_basis_state // 2
+    control_values = [1] * num_zero_ctrls + [0] * (size_basis_state - num_zero_ctrls)
+
+    multi_x = _map_to_bloq(
+        qops.MultiControlledX(wires=range(size_basis_state + 1), control_values=control_values)
+    )
+
+    basis_size = 2**size_basis_state
+    prob_matching_basis_states = num_basis_states / basis_size
+    num_permutes = round(num_basis_states * (1 - prob_matching_basis_states))
+    if num_permutes:
+        gate_types[cnot] = num_permutes * (size_basis_state // 2)  # average number of bits to flip
+        gate_types[multi_x] = 2 * num_permutes  # for compute and uncompute
+
+    return gate_types
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.state_preparations.QROMStatePreparation):
+    """Call graph for QROMStatePreparation"""
+    # From ResourceQROMStatePreparation
+
+    def _add_qrom_and_adjoint(gate_types, bitstrings, control_wires):
+        """Helper to create a QROM, count it and its adjoint."""
+        qrom_op = qtemps.QROM(
+            bitstrings=bitstrings,
+            target_wires=precision_wires,
+            control_wires=control_wires,
+            work_wires=work_wires,
+            clean=False,
+        )
+        gate_types[_map_to_bloq(qrom_op)] += 1
+        gate_types[_map_to_bloq(qops.adjoint(qrom_op))] += 1
+
+    gate_types = defaultdict(int, {})
+    positive_and_real = not any(c.imag != 0 or c.real < 0 for c in op.state_vector)
+
+    num_state_qubits = int(math.log2(len(op.state_vector)))
+    precision_wires = op.hyperparameters["precision_wires"]
+    input_wires = op.hyperparameters["input_wires"]
+    work_wires = op.hyperparameters["work_wires"]
+    num_precision_wires = len(precision_wires)
+
+    # Use helper for the first QROM
+    _add_qrom_and_adjoint(
+        gate_types, bitstrings=["0" * (num_precision_wires - 1) + "1"], control_wires=[]
+    )
+
+    zero_string = "0" * num_precision_wires
+    one_string = "0" * (num_precision_wires - 1) + "1" if num_precision_wires > 0 else ""
+
+    # Use helper inside the main loop
+    for i in range(1, num_state_qubits):
+        num_bit_flips = 2 ** (i - 1)
+        bitstrings = [zero_string] * num_bit_flips + [one_string] * num_bit_flips
+        _add_qrom_and_adjoint(gate_types, bitstrings, control_wires=input_wires[:i])
+
+    gate_types[_map_to_bloq(qops.CRY(0, wires=[0, 1]))] = num_precision_wires * num_state_qubits
+
+    # Use helper for the final conditional QROM
+    if not positive_and_real:
+        num_bit_flips = 2 ** (num_state_qubits - 1)
+        bitstrings = [zero_string] * num_bit_flips + [one_string] * num_bit_flips
+        _add_qrom_and_adjoint(gate_types, bitstrings, control_wires=input_wires)
+
+        gate_types[
+            _map_to_bloq(
+                qops.ctrl(
+                    qops.GlobalPhase((2 * np.pi), wires=input_wires[0]),
+                    control=0,
+                )
+            )
+        ] = num_precision_wires
+
+    return gate_types
+
+
+@_get_op_call_graph.register
+def _(op: qops.BasisState):
+    """Call graph for Basis State"""
+    gate_types = defaultdict(int, {})
+    gate_types[qt_gates.XGate()] = sum(op.parameters[0])
+
+    return gate_types
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.subroutines.QROM):
+    """Call graph for QROM"""
+
+    # From ResourceQROM
+    gate_types = defaultdict(int, {})
+    bitstrings = op.hyperparameters["bitstrings"]
+    num_bitstrings = len(bitstrings)
+
+    num_bit_flips = sum(bits.count("1") for bits in bitstrings)
+
+    num_work_wires = len(op.hyperparameters["work_wires"])
+    size_bitstring = len(op.hyperparameters["target_wires"])
+    num_control_wires = len(op.hyperparameters["control_wires"])
+    clean = op.hyperparameters["clean"]
+
+    if num_control_wires == 0:
+        gate_types[qt_gates.XGate()] = num_bit_flips
+        return gate_types
+
+    cnot = qt_gates.CNOT()
+    hadamard = qt_gates.Hadamard()
+    num_parallel_computations = (num_work_wires + size_bitstring) // size_bitstring
+
+    square_fact = math.floor(math.sqrt(num_bitstrings))  # use a square scheme for rows and cloumns
+    num_parallel_computations = min(num_parallel_computations, square_fact)
+
+    num_swap_wires = math.floor(math.log2(num_parallel_computations))
+    num_select_wires = math.ceil(math.log2(math.ceil(num_bitstrings / (2**num_swap_wires))))
+
+    swap_work_wires = (int(2**num_swap_wires) - 1) * size_bitstring
+    free_work_wires = num_work_wires - swap_work_wires
+
+    swap_clean_prefactor = 1
+    select_clean_prefactor = 1
+
+    if clean:
+        gate_types[hadamard] = 2 * size_bitstring
+        swap_clean_prefactor = 4
+        select_clean_prefactor = 2
+
+    # SELECT cost:
+    gate_types[cnot] = num_bit_flips  # each unitary in the select is just a CNOT
+
+    num_select_wires = int(num_select_wires)
+    multi_x = _map_to_bloq(
+        qops.MultiControlledX(
+            wires=range(num_select_wires + 1),
+            control_values=[True] * num_select_wires,
+            work_wires=range(num_select_wires + 1, num_select_wires + 1 + free_work_wires),
+        )
+    )
+
+    num_total_ctrl_possibilities = 2**num_select_wires
+    gate_types[multi_x] = select_clean_prefactor * (
+        2 * num_total_ctrl_possibilities  # two applications targetting the aux qubit
+    )
+    num_zero_controls = (2 * num_total_ctrl_possibilities * num_select_wires) // 2
+    gate_types[qt_gates.XGate()] = select_clean_prefactor * (
+        num_zero_controls * 2  # conjugate 0 controls on the multi-qubit x gates from above
+    )
+    # SWAP cost:
+    ctrl_swap = qt_gates.TwoBitCSwap()
+    gate_types[ctrl_swap] = swap_clean_prefactor * ((2**num_swap_wires) - 1) * size_bitstring
+
+    return gate_types
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.subroutines.QFT):
+    """Call graph for Quantum Fourier Transform"""
+
+    # From PL Decomposition
+    gate_types = defaultdict(int, {})
+    num_wires = len(op.wires)
+    gate_types[qt_gates.Hadamard()] = num_wires
+    gate_types[_map_to_bloq(qops.ControlledPhaseShift(1, [0, 1]))] = (
+        num_wires * (num_wires - 1) // 2
+    )
+    gate_types[qt_gates.TwoBitSwap()] = num_wires // 2
+    return gate_types
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.subroutines.QSVT):
+    """Call graph for Quantum Singular Value Transform"""
+
+    # From ResouceQSVT
+    gate_types = defaultdict(int, {})
+    UA = op.hyperparameters["UA"]
+    projectors = op.hyperparameters["projectors"]
+    num_projectors = len(projectors)
+
+    for proj_op in projectors[:-1]:
+        gate_types[_map_to_bloq(proj_op)] += 1
+
+    gate_types[_map_to_bloq(UA)] += num_projectors // 2
+    gate_types[_map_to_bloq(UA).adjoint()] += (num_projectors - 1) // 2
+    gate_types[_map_to_bloq(projectors[-1])] += 1
+
+    return gate_types
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.subroutines.Select):
+    """Call graph for Select"""
+
+    # From ResourceSelect
+    gate_types = defaultdict(int, {})
+    ops = op.hyperparameters["ops"]
+    cmpr_ops = [_map_to_bloq(op) for op in ops]
+
+    x = qt_gates.XGate()
+
+    num_ops = len(cmpr_ops)
+    num_ctrl_wires = int(np.ceil(np.log2(num_ops)))
+    num_total_ctrl_possibilities = 2**num_ctrl_wires  # 2^n
+
+    num_zero_controls = num_total_ctrl_possibilities // 2
+    gate_types[x] = num_zero_controls * 2  # conjugate 0 controls
+
+    for cmp_rep in cmpr_ops:
+        ctrl_op = cmp_rep.controlled(CtrlSpec(cvs=[1] * num_ctrl_wires))
+        if cmp_rep == qt_gates.XGate() and num_ctrl_wires == 1:
+            ctrl_op = qt_gates.CNOT()
+        gate_types[ctrl_op] += 1
+
+    return gate_types
+
+
+@_get_op_call_graph.register
+def _(op: qops.StatePrep):
+    """Call graph for StatePrep"""
+
+    # MottonenStatePrep
+    gate_types = defaultdict(int, {})
+    num_wires = len(op.wires)
+    rz = qt_gates.Rz(0)
+    cnot = qt_gates.CNOT()
+
+    r_count = 2 ** (num_wires + 2) - 5
+    cnot_count = 2 ** (num_wires + 2) - 4 * num_wires - 4
+
+    if r_count:
+        gate_types[rz] = r_count
+
+    if cnot_count:
+        gate_types[cnot] = cnot_count
+    return gate_types
+
+
+@_get_op_call_graph.register
+def _(op: qtemps.subroutines.ModExp):
+    """Call graph for ModExp"""
+
+    # From ResourceModExp
+    mod = op.hyperparameters["mod"]
+    num_work_wires = len(op.hyperparameters["work_wires"])
+    num_x_wires = len(op.hyperparameters["x_wires"])
+
+    mult_resources = {}
+    if mod == 2**num_x_wires:
+        num_aux_wires = num_x_wires
+        num_aux_swap = num_x_wires
+    else:
+        num_aux_wires = num_work_wires - 1
+        num_aux_swap = num_aux_wires - 1
+
+    qft = _map_to_bloq(qtemps.QFT(wires=range(num_aux_wires)), map_ops=False)
+    qft_dag = qft.adjoint()
+
+    sequence = _map_to_bloq(
+        qtemps.ControlledSequence(
+            qtemps.PhaseAdder(k=3, x_wires=range(1, num_x_wires + 1)), control=[0]
+        )
+    )
+    sequence_dag = sequence.adjoint()
+
+    cnot = qt_gates.CNOT()
+
+    mult_resources = {}
+    mult_resources[qft] = 2
+    mult_resources[qft_dag] = 2
+    mult_resources[sequence] = 1
+    mult_resources[sequence_dag] = 1
+    mult_resources[cnot] = min(num_x_wires, num_aux_swap)
+
+    gate_types = defaultdict(int, {})
+    ctrl_spec = CtrlSpec(cvs=[1])
+    for comp_rep in mult_resources:
+        new_rep = comp_rep.controlled(ctrl_spec)
+        if comp_rep == qt_gates.CNOT():
+            new_rep = qt_gates.Toffoli()
+        # cancel out QFTs from consecutive Multipliers
+        if hasattr(comp_rep, "op"):
+            if comp_rep.op.name == "QFT":
+                gate_types[new_rep] = 1
+        elif hasattr(comp_rep, "subbloq"):
+            if comp_rep.subbloq.op.name == "QFT":
+                gate_types[new_rep] = 1
+        else:
+            gate_types[new_rep] = mult_resources[comp_rep] * ((2**num_x_wires) - 1)
+
+    return gate_types
+
+
+@singledispatch
+def _map_to_bloq(op, map_ops=True, custom_mapping=None, **kwargs):
+    """Map PennyLane operators to Qualtran Bloqs. Operators with direct equivalents are directly
+    mapped to their Qualtran equivalent even if ``map_ops`` is set to ``False``. Other operators are
+    given a smart default mapping. When given a ``custom_mapping``, the custom mapping is used."""
+    if not isinstance(op, Operator):
+        return ToBloq(op, map_ops=map_ops, custom_mapping=custom_mapping, **kwargs)
+
+    if custom_mapping is not None:
+        return custom_mapping[op]
+
+    return ToBloq(op, map_ops=map_ops, **kwargs)
+
+
+def _handle_custom_map(op, map_ops, custom_mapping, **kwargs):
+    """
+    Handles the custom mapping and wrapping logic
+
+    Args:
+        op (Operation): a PennyLane operator to be converted to a Qualtran Bloq.
+        map_ops (bool): Whether to map operations to a Qualtran Bloq. Operations are wrapped
+            as a ``ToBloq`` when False. Default is True.
+        custom_mapping (dict): Dictionary to specify a mapping between a PennyLane operator and a
+            Qualtran Bloq. Default is None.
+
+    Returns:
+        Optional[`Bloq`]: A ``ToBloq`` or the ``Bloq`` defined in the custom mapping. ``None`` if
+            map_ops is True but no custom mapping is found.
+    """
+
+    if not map_ops:
+        return ToBloq(op, **kwargs)
+
+    if custom_mapping is not None and op in custom_mapping:
+        return custom_mapping[op]
+
+    return None
+
+
+# pylint: disable=import-outside-toplevel
+@_map_to_bloq.register
+def _(
+    op: qtemps.subroutines.qpe.QuantumPhaseEstimation,
+    map_ops=True,
+    custom_mapping=None,
+    **kwargs,
+):
+    from qualtran.bloqs.phase_estimation import RectangularWindowState
+    from qualtran.bloqs.phase_estimation.text_book_qpe import TextbookQPE
+
+    mapped_op = _handle_custom_map(op, map_ops, custom_mapping, **kwargs)
+    if mapped_op is not None:
+        return mapped_op
+
+    return TextbookQPE(
+        unitary=_map_to_bloq(op.hyperparameters["unitary"]),
+        ctrl_state_prep=RectangularWindowState(len(op.hyperparameters["estimation_wires"])),
+    )
+
+
+# pylint: disable=import-outside-toplevel
+@_map_to_bloq.register
+def _(op: qtemps.subroutines.QFT, custom_mapping=None, map_ops=True, **kwargs):
+    """Mapping for QFT, which maps to ``qt.QFTTextBook`` by default"""
+    from qualtran.bloqs.qft import QFTTextBook
+
+    mapped_op = _handle_custom_map(op, map_ops, custom_mapping, **kwargs)
+    if mapped_op is not None:
+        return mapped_op
+
+    return QFTTextBook(len(op.wires))
+
+
+# pylint: disable=import-outside-toplevel
+@_map_to_bloq.register
+def _(op: qtemps.subroutines.QROM, map_ops=True, custom_mapping=None, **kwargs):
+    """Mapping for QROM that defaults to either ``QROAMClean`` or ``SelectSwapQROM``"""
+    from qualtran.bloqs.data_loading.qroam_clean import QROAMClean
+    from qualtran.bloqs.data_loading.select_swap_qrom import SelectSwapQROM
+
+    mapped_op = _handle_custom_map(op, map_ops, custom_mapping, **kwargs)
+    if mapped_op is not None:
+        return mapped_op
+
+    data = np.array([int(b, 2) for b in op.bitstrings])
+    if op.clean:
+        return QROAMClean.build_from_data(data)
+
+    return SelectSwapQROM.build_from_data(data)
+
+
+# pylint: disable=import-outside-toplevel
+@_map_to_bloq.register
+def _(op: qtemps.subroutines.ModExp, map_ops=True, custom_mapping=None, **kwargs):
+    """Mapping for ``ModExp``"""
+    from qualtran.bloqs.cryptography.rsa import ModExp
+
+    mapped_op = _handle_custom_map(op, map_ops, custom_mapping, **kwargs)
+    if mapped_op is not None:
+        return mapped_op
+
+    return ModExp(
+        base=op.hyperparameters["base"],
+        mod=op.hyperparameters["mod"],
+        exp_bitsize=len(op.hyperparameters["x_wires"]),
+        x_bitsize=len(op.hyperparameters["output_wires"]),
+    )
+
+
+def _disable_custom_mapping(func):
+    """Decorator to disable custom mapping and raise error for atomic gates."""
+
+    @wraps(func)
+    def wrapper(op, **kwargs):
+        if kwargs.get("custom_mapping") and op in kwargs.get("custom_mapping", {}):
+            raise ValueError(
+                "Custom mappings are not possible for basic operations. We suggest replacing basic operations "
+                "such as X, Y, or other gates with direct Qualtran equivalents, before or after mapping to Qualtran."
+            )
+        return func(op, **kwargs)
+
+    return wrapper
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.GlobalPhase, **kwargs):
+    return qt_gates.GlobalPhase(exponent=op.data[0] / np.pi)
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.Hadamard, **kwargs):
+    return qt_gates.Hadamard()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.Identity, **kwargs):
+    return qt_gates.Identity()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.RX, **kwargs):
+    return qt_gates.Rx(angle=float(op.data[0]))
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.RY, **kwargs):
+    return qt_gates.Ry(angle=float(op.data[0]))
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.RZ, **kwargs):
+    return qt_gates.Rz(angle=float(op.data[0]))
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.S, **kwargs):
+    return qt_gates.SGate()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.SWAP, **kwargs):
+    return qt_gates.TwoBitSwap()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.CSWAP, **kwargs):
+    return qt_gates.TwoBitCSwap()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.T, **kwargs):
+    return qt_gates.TGate()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.X, **kwargs):
+    return qt_gates.XGate()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.Y, **kwargs):
+    return qt_gates.YGate()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.CY, **kwargs):
+    return qt_gates.CYGate()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.Z, **kwargs):
+    return qt_gates.ZGate()
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qops.CZ, **kwargs):
+    return qt_gates.CZ()
+
+
+@_map_to_bloq.register
+def _(op: qops.Adjoint, map_ops=True, custom_mapping=None, **kwargs):
+    return _map_to_bloq(op.base, custom_mapping=custom_mapping, map_ops=map_ops, **kwargs).adjoint()
+
+
+@_map_to_bloq.register
+def _(op: qops.Controlled, map_ops=True, custom_mapping=None, **kwargs):
+    if isinstance(op, qops.CNOT):
+        return qt_gates.CNOT()
+
+    ctrl_spec = CtrlSpec(cvs=[int(v) for v in op.control_values])
+    return _map_to_bloq(
+        op.base, map_ops=map_ops, custom_mapping=custom_mapping, **kwargs
+    ).controlled(ctrl_spec)
+
+
+@_map_to_bloq.register
+@_disable_custom_mapping
+def _(op: qmeas.MeasurementProcess, **kwargs):
+    return None
 
 
-@lru_cache
 def _get_to_pl_op():
     @singledispatch
     def _to_pl_op(bloq, wires):
@@ -120,13 +759,13 @@ def _get_to_pl_op():
     return _to_pl_op
 
 
-def bloq_registers(bloq):
+def bloq_registers(bloq: "qt.Bloq"):
     """Reads a `Qualtran Bloq <https://qualtran.readthedocs.io/en/latest/bloqs/index.html#bloqs-library>`_
     signature and returns a dictionary mapping the Bloq's register names to :class:`~.Wires`.
 
     .. note::
-        This function requires the latest version of Qualtran. We recommend installing the main
-        branch via ``pip``:
+        This function requires the latest version of Qualtran. We recommend installing the latest
+        release via ``pip``:
 
         .. code-block:: console
 
@@ -144,7 +783,7 @@ def bloq_registers(bloq):
 
     Returns:
         dict: A dictionary mapping the names of the Bloq's registers to :class:`~.Wires`
-            objects with the same lengths as the bitsizes of their respective registers.
+        objects with the same lengths as the bitsizes of their respective registers.
 
     Raises:
         TypeError: bloq must be an instance of ``Bloq``.
@@ -215,8 +854,8 @@ class FromBloq(Operation):
     as a PennyLane :class:`~.Operation`.
 
     .. note::
-        This class requires the latest version of Qualtran. We recommend installing the main
-        branch via ``pip``:
+        This class requires the latest version of Qualtran. We recommend installing the latest
+        release via ``pip``:
 
         .. code-block:: console
 
@@ -418,3 +1057,403 @@ class FromBloq(Operation):
             raise MatrixUndefinedError
 
         return matrix
+
+
+class _QReg:
+    """Used as a container for qubits that form a `Register` of a given bitsize. This is a modified
+    version of `_QReg <https://github.com/quantumlib/Qualtran/blob/main/qualtran/cirq_interop/_cirq_to_bloq.py>`_
+    found in Qualtran as well.
+
+    Each instance of `_QReg` would correspond to a `Soquet` in Bloqs and represents an opaque collection
+    of qubits that together form a quantum register.
+    """
+
+    def __init__(self, qubits: Tuple["cirq.Qid", ...], dtype: "qt.QDType"):
+        if isinstance(qubits, cirq.Qid):
+            self.qubits = (qubits,)
+        else:
+            self.qubits = tuple(qubits)
+
+        self.dtype = dtype
+        self._initialized = True
+
+    def __setattr__(self, name, value):
+        """Makes the instance immutable after initialization."""
+        if getattr(self, "_initialized", False):
+            raise AttributeError(
+                f"Cannot set attribute '{name}'. Instances of _QReg are immutable."
+            )
+        super().__setattr__(name, value)
+
+    def __repr__(self) -> str:
+        """Provides a developer-friendly string representation."""
+        return f"_QReg(qubits={self.qubits!r}, dtype={self.dtype!r})"
+
+    # Override the __eq__ and __hash__ functions to handle single qubit registers
+    # that are functionally the same but have different dtypes.
+    def __eq__(self, other) -> bool:
+        if not isinstance(other, _QReg):
+            return False
+        return self.qubits == other.qubits
+
+    def __hash__(self):
+        return hash(self.qubits)
+
+
+def _ensure_in_reg_exists(
+    bb: "qt.BloqBuilder",
+    in_reg: "_QReg",
+    qreg_to_qvar: Dict["_QReg", "qt.Soquet"],
+) -> None:
+    """Modified function from the Qualtran-Cirq interop module to ensure `qreg_to_qvar[in_reg]`
+    exists. If `in_reg` is not found in `qreg_to_qvar`, that means that the input qubit register
+    is a multi-qubit register. All in_regs should be single qubit registers, so this would be a
+    bug, and an AssertionError is raised. To capture control flow, multi-qubit registers will be
+    allowed, and we will remove the AssertionError and use Split and Join operations as needed.
+
+    Args:
+        bb (qt.BloqBuilder): an instance of a Qualtran BloqBuilder
+        in_reg (_QReg): a container for qubits that form a Register of a given bitsize
+        qreg_to_qvar (Dict[_QReg, qt.Soquet]): a dictionary of _QRegs that corresponds to Soquets
+
+    Raises:
+        AssertionError: `in_reg` was not found in `qreg_to_qvar`, meaning there exists multi-qubit
+            registers that we do not support at the moment
+    """
+
+    all_mapped_qubits = {q for qreg in qreg_to_qvar for q in qreg.qubits}
+    qubits_to_allocate = [q for q in in_reg.qubits if q not in all_mapped_qubits]
+    if qubits_to_allocate:
+        n_alloc = len(qubits_to_allocate)
+        qreg_to_qvar[
+            _QReg(qubits_to_allocate, dtype=qt.QBit() if n_alloc == 1 else qt.QAny(n_alloc))
+        ] = bb.allocate(n_alloc)
+
+    # if in_reg not in qreg_to_qvar: splits & joins needed, which shouldn't be the case
+    assert in_reg in qreg_to_qvar, f"Input register {in_reg} not found, suggesting a bug"
+
+
+def _gather_input_soqs(bb: "qt.BloqBuilder", op_quregs, qreg_to_qvar):
+    """Modified function from Qualtran-Cirq interop module that collects input Soquets.
+
+    Args:
+        bb (qt.BloqBuilder): an instance of a Qualtran BloqBuilder
+        op_quregs (Dict[str, _QRegs]): a dict of register names that corresponds to _QRegs
+        qreg_to_qvar (Dict[str, qt.Soquet]): a dict of register names that corresponds to input Soquets
+
+    Returns:
+        dict: in_reg was not found in qreg_to_qvar
+    """
+    qvars_in = {}
+    for reg_name, quregs in op_quregs.items():
+        flat_soqs: List[qt.Soquet] = []
+        for qureg in quregs.flatten():
+            _ensure_in_reg_exists(bb, qureg, qreg_to_qvar)
+            flat_soqs.append(qreg_to_qvar[qureg])
+        qvars_in[reg_name] = np.array(flat_soqs).reshape(quregs.shape)
+    return qvars_in
+
+
+class ToBloq(Bloq):  # pylint:disable=useless-object-inheritance (Inherit qt.Bloq optionally)
+    r"""
+    An adapter to convert a PennyLane :class:`~.QNode`, ``Qfunc``, or :class:`~.Operation` to a
+    `Qualtran Bloq <https://qualtran.readthedocs.io/en/latest/bloqs/index.html#bloqs-library>`__.
+
+    .. note::
+        This class requires the latest version of Qualtran. We recommend installing the latest
+        release via ``pip``:
+
+        .. code-block:: console
+
+            pip install qualtran
+
+    Args:
+        op (QNode| Qfunc | Operation): a PennyLane ``QNode``, ``Qfunc``, or operator to be wrapped
+            as a Qualtran Bloq.
+        map_ops (bool): Whether to map operations to a Qualtran Bloq. Operations are wrapped
+            as a ``ToBloq`` when ``False``. Default is ``True``.
+        custom_mapping (dict): Dictionary to specify a mapping between a PennyLane operator and a
+            Qualtran Bloq. A default mapping is used if not defined.
+
+    Raises:
+        TypeError: operator must be an instance of :class:`~.Operation`.
+
+    .. seealso:: :func:`~.to_bloq`
+
+    **Example**
+
+    This example shows how to use ``qml.ToBloq``:
+
+    >>> from qualtran.resource_counting.generalizers import generalize_rotation_angle
+    >>> op = qml.QuantumPhaseEstimation(
+    ...     qml.RX(0.2, wires=[0]), estimation_wires=[1, 2]
+    ... )
+    >>> op_as_bloq = qml.ToBloq(op)
+    >>> graph, sigma = op_as_bloq.call_graph(generalize_rotation_angle)
+    >>> sigma
+    {Hadamard(): 4,
+    Controlled(subbloq=Rx(angle=0.2, eps=1e-11), ctrl_spec=CtrlSpec(qdtypes=(QBit(),), cvs=(array(1),))): 3,
+    TwoBitSwap(): 1,
+    CNOT(): 2,
+    ZPowGate(exponent=\phi, eps=5e-12): 2,
+    ZPowGate(exponent=\phi, eps=1e-11): 1}
+    """
+
+    def __init__(self, op, map_ops=False, custom_mapping=None, **kwargs):
+        if not qualtran:
+            raise ImportError(
+                "Optional dependency 'qualtran' is required "
+                "for ToBloq functionality but is not installed. Try `pip install qualtran`."
+            )
+
+        if not isinstance(op, Operator) and not isinstance(op, QNode) and not callable(op):
+            raise TypeError(
+                f"Input must be either an instance of {Operator}, {QNode} or a quantum function."
+            )
+
+        self.op = op
+        self.map_ops = map_ops
+        self.custom_mapping = custom_mapping
+        self._kwargs = kwargs
+        super().__init__()
+
+    @cached_property
+    def signature(self) -> "qt.Signature":
+        """Compute and return Qualtran signature for given op or QNode."""
+        if isinstance(self.op, QNode):
+            self.op.name = "QNode"
+            num_wires = len(construct_tape(self.op)(**self._kwargs).wires)
+        elif isinstance(self.op, Operator):
+            num_wires = len(self.op.wires)
+        else:
+            num_wires = len(make_qscript(self.op)(**self._kwargs).wires)
+        return qt.Signature([qt.Register("qubits", qt.QBit(), shape=num_wires)])
+
+    def decompose_bloq(self):  # pylint:disable=too-many-branches
+        """Decompose the bloq using the op's decomposition or the tape of the QNode"""
+        try:
+            if isinstance(self.op, QNode):
+                tape = construct_tape(self.op)(**self._kwargs)
+                ops = tape.circuit
+                all_wires = list(tape.wires)
+            elif isinstance(self.op, Operator):
+                ops = self.op.decomposition()
+                all_wires = list(self.op.wires)
+            else:
+                tape = make_qscript(self.op)(**self._kwargs)
+                ops = tape.operations
+                all_wires = list(tape.wires)
+
+            signature = self.signature
+            in_quregs = out_quregs = {"qubits": np.array(all_wires).reshape(len(all_wires), 1)}
+
+            in_key = list(in_quregs.keys())[0]
+            out_key = list(out_quregs.keys())[0]
+
+            in_quregs = {
+                in_key: np.apply_along_axis(
+                    _QReg, -1, in_quregs[in_key], signature.get_left(in_key).dtype
+                )
+            }
+            out_quregs = {
+                out_key: np.apply_along_axis(
+                    _QReg, -1, out_quregs[out_key], signature.get_right(out_key).dtype
+                )
+            }
+
+            bb, initial_soqs = qt.BloqBuilder.from_signature(signature, add_registers_allowed=False)
+
+            # `signature.lefts()` can be thought of as input qubits. For our purposes LEFT and
+            # RIGHT signatures will in most cases match since there are no allocated & freed
+            # qubits. Here, qreg_to_qvar is a map between a register and a Soquet. This serves
+            # as the foundation to wire up the rest of the bloqs.
+            qreg_to_qvar = {}
+            for reg in signature.lefts():
+                assert reg.name in in_quregs
+                soqs = initial_soqs[reg.name]
+                assert in_quregs[reg.name].shape == soqs.shape
+                qreg_to_qvar |= zip(in_quregs[reg.name].flatten(), soqs.flatten())
+
+            # Add each operation to the composite Bloq.
+            for op in ops:
+                bloq = _map_to_bloq(
+                    op, map_ops=self.map_ops, custom_mapping=self.custom_mapping, **self._kwargs
+                )
+                if bloq is None:
+                    continue
+
+                if bloq.signature == qt.Signature([]):
+                    bb.add(bloq)
+                    continue
+
+                reg_dtypes = [r.dtype for r in bloq.signature]
+                # Find input / output registers.
+                all_op_quregs = {
+                    k: np.apply_along_axis(_QReg, -1, *(v, reg_dtypes[i]))  # type: ignore
+                    for i, (k, v) in enumerate(split_qubits(bloq.signature, op.wires).items())
+                }
+
+                in_op_quregs = {reg.name: all_op_quregs[reg.name] for reg in bloq.signature.lefts()}
+                # Find input Soquets, by potentially allocating new Bloq registers corresponding to
+                # input `in_quregs` and updating the `qreg_to_qvar` mapping.
+                qvars_in = _gather_input_soqs(bb, in_op_quregs, qreg_to_qvar)
+
+                # Add Bloq to the `CompositeBloq` compute graph and get corresponding output Soquets.
+                qvars_out = bb.add_d(bloq, **qvars_in)
+
+                # Update `qreg_to_qvar` mapping using output soquets `qvars_out`.
+                for reg in bloq.signature:
+                    # all_op_quregs should exist for both LEFT & RIGHT registers.
+                    assert reg.name in all_op_quregs
+                    quregs = all_op_quregs[reg.name]
+                    if reg.side != qt.Side.LEFT:
+                        assert quregs.shape == np.array(qvars_out[reg.name]).shape
+                        qreg_to_qvar |= zip(
+                            quregs.flatten(), np.array(qvars_out[reg.name]).flatten()
+                        )
+
+            # Combine Soquets to match the right signature.
+            final_soqs_dict = _gather_input_soqs(
+                bb,
+                {reg.name: out_quregs[reg.name] for reg in signature.rights()},
+                qreg_to_qvar,
+            )
+            final_soqs_set = set(soq for soqs in final_soqs_dict.values() for soq in soqs.flatten())
+            # Free all dangling Soquets which are not part of the final soquets set.
+            for qvar in qreg_to_qvar.values():
+                if qvar not in final_soqs_set:
+                    bb.free(qvar)
+
+            cbloq = bb.finalize(**final_soqs_dict)
+            return cbloq
+        except DecompositionUndefinedError as undefined_decomposition:
+            raise qt.DecomposeNotImplementedError from undefined_decomposition
+
+    def build_call_graph(self, ssa):
+        """Build Qualtran call graph with defined call graph if available, otherwise build
+        said call graph with the decomposition"""
+        call_graph = _get_op_call_graph(self.op)
+        if call_graph:
+            return call_graph
+
+        return self.decompose_bloq().build_call_graph(ssa)
+
+    def __repr__(self):
+        if isinstance(self.op, QNode):
+            return "ToBloq(QNode)"
+        if isinstance(self.op, Operation):
+            return f"ToBloq({self.op.name})"
+        return "ToBloq(Qfunc)"
+
+    def __eq__(self, other):
+        if type(other) is type(self):
+            return self.op == other.op
+        return False
+
+    def __hash__(self):
+        return hash(self.op)
+
+    def __str__(self):
+        if hasattr(self.op, "name"):
+            return f"PL{self.op.name}"
+        return "PLQfunc"
+
+
+def to_bloq(circuit, map_ops: bool = True, custom_mapping: dict = None, **kwargs):
+    """
+    Converts a PennyLane :class:`~.QNode`, ``Qfunc``, or :class:`~.Operation` to the corresponding `Qualtran Bloq <https://qualtran.readthedocs.io/en/latest/bloqs/index.html#bloqs-library>`__.
+
+    .. note::
+        This class requires the latest version of Qualtran. We recommend installing the latest
+        release via ``pip``:
+
+        .. code-block:: console
+
+            pip install qualtran
+
+    Args:
+        circuit (QNode| Qfunc | Operation): a PennyLane ``QNode``, ``Qfunc``, or operator to be wrapped
+            as a Qualtran Bloq.
+        map_ops (bool): Whether to map operations to a Qualtran Bloq. Operations are wrapped
+            as a ``ToBloq`` when ``False``. Default is ``True``.
+        custom_mapping (dict): Dictionary to specify a mapping between a PennyLane operator and a
+            Qualtran Bloq. A default mapping is used if not defined.
+
+    Returns:
+        Bloq: The Qualtran Bloq that corresponds to the given circuit or :class:`~.Operation` and
+        options.
+
+    .. seealso:: :class:`~.ToBloq`
+
+    **Example**
+
+    This example shows how to use ``qml.to_bloq``:
+
+    >>> from qualtran.resource_counting.generalizers import generalize_rotation_angle
+    >>> op = qml.QuantumPhaseEstimation(
+    ...     qml.RX(0.2, wires=[0]), estimation_wires=[1, 2]
+    ... )
+    >>> op_as_bloq = qml.to_bloq(op)
+    >>> graph, sigma = op_as_bloq.call_graph(generalize_rotation_angle)
+    >>> sigma
+    {Allocate(dtype=QFxp(bitsize=2, num_frac=2, signed=False), dirty=False): 1,
+    Hadamard(): 4,
+    Controlled(subbloq=Rx(angle=0.2, eps=1e-11), ctrl_spec=CtrlSpec(qdtypes=(QBit(),), cvs=(array(1),))): 3,
+    And(cv1=1, cv2=1, uncompute=True): 1,
+    And(cv1=1, cv2=1, uncompute=False): 1,
+    ZPowGate(exponent=\\phi, eps=1e-10): 1,
+    TwoBitSwap(): 1}
+
+    .. details::
+        :title: Usage Details
+
+        Some PennyLane operators don't have a direct equivalent in Qualtran. For example, in Qualtran, there
+        are many varieties of Quantum Phase Estimation. When ``qml.to_bloq`` is called on
+        :class:`~pennylane.QuantumPhaseEstimation`, a smart default is chosen.
+
+        >>> qml.to_bloq(qml.QuantumPhaseEstimation(
+        ...     unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)
+        ... ))
+        TextbookQPE(unitary=Rx(angle=0.1, eps=1e-11), ctrl_state_prep=RectangularWindowState(bitsize=4), qft_inv=Adjoint(subbloq=QFTTextBook(bitsize=4, with_reverse=True)))
+
+        Note that the chosen Qualtran Bloq may not be an exact equivalent. If an exact
+        equivalent is needed, we recommend setting ``map_ops`` to ``False``.
+        This will wrap the input PennyLane operator as a Qualtran Bloq, enabling Qualtran functions
+        such as ``decompose_bloq`` or ``call_graph``, but maintaining the PennyLane decomposition definition of the operator.
+
+        >>> qml.to_bloq(qml.QuantumPhaseEstimation(
+        ...     unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)
+        ... ), map_ops=False)
+        ToBloq(QuantumPhaseEstimation)
+
+
+        Alternatively, users can provide a custom mapping that maps a PennyLane operator to a
+        specific Qualtran Bloq. It is recommended to map operators at the high level, rather than
+        attempt to map operators that appear in the operator's decomposition.
+
+        >>> from qualtran.bloqs.phase_estimation import TextbookQPE
+        >>> from qualtran.bloqs.phase_estimation.lp_resource_state import LPResourceState
+        >>> op = qml.QuantumPhaseEstimation(
+        ...         unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)
+        ...     )
+        >>> custom_mapping = {
+        ...     op : TextbookQPE(
+        ...         unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+        ...         ctrl_state_prep=LPResourceState(4),
+        ...     )
+        ... }
+        >>> qml.to_bloq(op, custom_mapping=custom_mapping)
+        TextbookQPE(unitary=Rx(angle=0.1, eps=1e-11), ctrl_state_prep=LPResourceState(bitsize=4), qft_inv=Adjoint(subbloq=QFTTextBook(bitsize=4, with_reverse=True)))
+
+    """
+
+    if not qualtran:
+        raise ImportError(
+            "The `to_bloq` function requires Qualtran to be installed. You can install"
+            "qualtran via: pip install qualtran."
+        )
+
+    if map_ops and custom_mapping:
+        return _map_to_bloq(circuit, map_ops=True, custom_mapping=custom_mapping, **kwargs)
+
+    return _map_to_bloq(circuit, map_ops=map_ops, **kwargs)


Test patch:
diff --git a/tests/io/test_qualtran_io.py b/tests/io/test_qualtran_io.py
index 35a06c797..5343aa5fd 100644
--- a/tests/io/test_qualtran_io.py
+++ b/tests/io/test_qualtran_io.py
@@ -18,7 +18,7 @@ import numpy as np
 import pytest
 
 import pennylane as qml
-from pennylane.io.qualtran_io import _get_to_pl_op
+from pennylane.io.qualtran_io import _get_op_call_graph, _get_to_pl_op, _map_to_bloq, _QReg
 from pennylane.operation import DecompositionUndefinedError
 
 
@@ -28,6 +28,40 @@ def skip_if_no_pl_qualtran_support():
     pytest.importorskip("qualtran")
 
 
+def test_to_bloq_error():
+    """Test import error message when ToBloq() is instantiated without qualtran installed"""
+    try:
+        import qualtran  # pylint: disable=unused-import
+    except (ModuleNotFoundError, ImportError):
+        with pytest.raises(ImportError, match="Optional dependency"):
+            qml.ToBloq(qml.H(0))
+
+        with pytest.raises(ImportError, match="The `to_bloq` function requires Qualtran "):
+            qml.to_bloq(qml.H(0))
+
+
+@pytest.mark.external
+@pytest.mark.usefixtures("skip_if_no_pl_qualtran_support")
+@pytest.fixture
+def qubits():
+    """Provides cirq.LineQubit instances for tests."""
+    import cirq
+
+    return cirq.LineQubit(0), cirq.LineQubit(1)
+
+
+@pytest.mark.external
+@pytest.mark.usefixtures("skip_if_no_pl_qualtran_support")
+@pytest.fixture
+def dtypes():
+    """Provides qualtran QDType instances for tests."""
+    from qualtran import QBit, QUInt
+
+    # A single QBit and a single-bit QUInt are different types
+    # but should be equivalent in a 1-qubit _QReg comparison.
+    return QBit(), QUInt(bitsize=1)
+
+
 @pytest.mark.external
 @pytest.mark.usefixtures("skip_if_no_pl_qualtran_support")
 class TestFromBloq:
@@ -333,3 +367,859 @@ class TestFromBloq:
         actual = qml.bloq_registers(circuit_bloq)
 
         assert actual == expected
+
+
+@pytest.mark.external
+@pytest.mark.usefixtures("skip_if_no_pl_qualtran_support")
+class TestToBloq:
+    """Test that ToBloq and to_bloq accurately wraps or maps Bloqs."""
+
+    def test_to_bloq_init(self):
+        """Tests that ToBloq's __init__() functions as intended"""
+
+        dev = qml.device("default.qubit")
+
+        @qml.qnode(dev)
+        def circuit():
+            qml.H(0)
+
+        assert qml.ToBloq(qml.Hadamard(0)).__repr__() == "ToBloq(Hadamard)"
+        assert qml.ToBloq(circuit).__repr__() == "ToBloq(QNode)"
+        assert qml.ToBloq(qml.H(0)).__str__() == "PLHadamard"
+        with pytest.raises(TypeError, match="Input must be either an instance of"):
+            qml.ToBloq("123")
+
+    def test_equivalence(self):
+        """Tests that ToBloq's __eq__ functions as expected"""
+
+        assert qml.ToBloq(qml.H(0)) == qml.ToBloq(qml.H(0))
+        assert qml.ToBloq(qml.H(0)) != qml.ToBloq(qml.H(1))
+        assert qml.ToBloq(qml.H(0)) != "Hadamard"
+
+    def test_allocate_and_free(self):
+        """Tests that ToBloq functions on a FromBloq that has ghost wires"""
+        from qualtran._infra.data_types import QAny, QBit
+        from qualtran.bloqs.basic_gates import CZPowGate
+        from qualtran.bloqs.bookkeeping import Allocate, Free
+
+        assert (
+            qml.to_bloq(qml.FromBloq(CZPowGate(0.468, eps=1e-11), wires=[0, 1])).call_graph()[1][
+                Allocate(QAny(bitsize=1))
+            ]
+            == 3
+        )
+        assert (
+            qml.to_bloq(qml.FromBloq(CZPowGate(0.468, eps=1e-11), wires=[0, 1])).call_graph()[1][
+                Free(QBit())
+            ]
+            == 3
+        )
+
+    def test_to_bloq(self):
+        """Tests that to_bloq functions as intended for simple circuits and gates"""
+
+        from qualtran.bloqs.basic_gates import Hadamard
+
+        dev = qml.device("default.qubit")
+
+        @qml.qnode(dev)
+        def circuit():
+            qml.H(0)
+            return (
+                qml.expval(qml.Y(0)),
+                qml.probs(op=qml.X(0)),
+                qml.state(),
+                qml.sample(qml.X(0)),
+                qml.var(qml.X(0)),
+                qml.counts(qml.X(0)),
+            )
+
+        def qfunc():
+            qml.H(0)
+
+        assert qml.to_bloq(qml.Hadamard(0)) == Hadamard()
+        assert qml.to_bloq(circuit).__repr__() == "ToBloq(QNode)"
+        assert qml.to_bloq(qfunc).__repr__() == "ToBloq(Qfunc)"
+        assert qml.to_bloq(qfunc).__str__() == "PLQfunc"
+        assert qml.to_bloq(qml.Hadamard(0), map_ops=False).__repr__() == "Hadamard()"
+        assert qml.to_bloq(circuit).call_graph()[1] == {Hadamard(): 1}
+        assert qml.to_bloq(qfunc).call_graph()[1] == {Hadamard(): 1}
+
+        with pytest.raises(
+            ValueError, match="Custom mappings are not possible for basic operations"
+        ):
+            qml.to_bloq(qml.X(0), custom_mapping={qml.X(0): qml.Y(0)})
+
+    def test_to_bloq_circuits(self):
+        """Tests that to_bloq functions as intended for complex circuits"""
+
+        from qualtran.bloqs.basic_gates import CNOT, Hadamard
+
+        dev = qml.device("default.qubit", wires=6)
+
+        @qml.qnode(dev)
+        def circuit():
+            qml.H(0)
+            qml.QuantumPhaseEstimation(unitary=qml.RX(0.1, wires=5), estimation_wires=range(5))
+
+        mapped_circuit = qml.to_bloq(circuit)
+        mapped_circuit_cg = mapped_circuit.call_graph()[1]
+        custom_mapped_circuit = qml.to_bloq(
+            circuit,
+            custom_mapping={
+                qml.QuantumPhaseEstimation(
+                    unitary=qml.RX(0.1, wires=5), estimation_wires=range(5)
+                ): Hadamard()
+            },
+        )
+        custom_mapped_circuit_cg = custom_mapped_circuit.call_graph()[1]
+        wrapped_circuit = qml.to_bloq(circuit, map_ops=False)
+        wrapped_circuit_cg = wrapped_circuit.call_graph()[1]
+
+        assert mapped_circuit_cg[Hadamard()] == 11
+        assert wrapped_circuit_cg[Hadamard()] == 11
+        assert custom_mapped_circuit_cg[Hadamard()] == 2
+        assert CNOT() not in mapped_circuit_cg
+        assert wrapped_circuit_cg[CNOT()] == 20
+
+    def test_from_bloq_to_bloq(self):
+        """Tests that FromBloq and to_bloq functions as intended"""
+
+        qpe_op = qml.QuantumPhaseEstimation(
+            unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)
+        )
+        qpe_bloq = qml.to_bloq(qpe_op, map_ops=False)
+
+        decomp_ops = qml.FromBloq(qpe_bloq, wires=range(5)).decomposition()
+        expected_decomp_ops = qpe_op.decomposition()
+        assert decomp_ops == [
+            qml.H(1),
+            qml.H(2),
+            qml.H(3),
+            qml.H(4),
+            qml.FromBloq(_map_to_bloq(expected_decomp_ops[4]), wires=[1, 0]),
+            qml.FromBloq(_map_to_bloq(expected_decomp_ops[5]), wires=[2, 0]),
+            qml.FromBloq(_map_to_bloq(expected_decomp_ops[6]), wires=[3, 0]),
+            qml.FromBloq(_map_to_bloq(expected_decomp_ops[7]), wires=[4, 0]),
+            qml.FromBloq(_map_to_bloq(expected_decomp_ops[8], map_ops=False), wires=range(1, 5)),
+        ]
+
+    def test_circuit_to_bloq_kwargs(self):
+        """Tests that to_bloq functions as intended for circuits with kwargs"""
+
+        from qualtran.bloqs.basic_gates import GlobalPhase, Rx
+
+        dev = qml.device("default.qubit")
+
+        @qml.qnode(dev)
+        def circuit(angle):
+            qml.RX(phi=angle, wires=[0])
+            qml.GlobalPhase(angle)
+
+        assert qml.to_bloq(circuit, angle=0).call_graph()[1] == {
+            Rx(angle=0.0, eps=1e-11): 1,
+            GlobalPhase(exponent=0): 1,
+        }
+        with pytest.raises(TypeError):
+            qml.to_bloq(circuit).call_graph()
+
+        assert qml.to_bloq(circuit, map_ops=False, angle=0).call_graph()[1] == {
+            Rx(angle=0.0, eps=1e-11): 1,
+            GlobalPhase(exponent=0): 1,
+        }
+
+    def test_decomposition_undefined_error(self):
+        """Tests that DecomposeTypeError is raised when the input op has no decomposition"""
+        import qualtran as qt
+
+        with pytest.raises(qt.DecomposeTypeError):
+            qml.to_bloq(qml.RZ(phi=0.3, wires=[0]), map_ops=False).decompose_bloq()
+
+    def test_call_graph(self):
+        """Tests that build_call_graph calls build_call_graph as expected"""
+        from qualtran.resource_counting import SympySymbolAllocator as ssa
+
+        cg = qml.to_bloq(
+            qml.QuantumPhaseEstimation(unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)),
+            False,
+        ).build_call_graph(ssa=ssa())
+
+        assert cg == {
+            qml.to_bloq(qml.Hadamard(0), True): 4,
+            qml.to_bloq(qml.ctrl(qml.RX(0.1, wires=0), control=[1]), True): 15,
+            qml.to_bloq(qml.adjoint(qml.QFT(wires=range(1, 5))), False): 1,
+        }
+
+    def test_map_to_bloq(self):
+        """Tests that _map_to_bloq produces the correct Qualtran equivalent"""
+        from qualtran.bloqs.basic_gates import (
+            CNOT,
+            CZ,
+            CYGate,
+            GlobalPhase,
+            Identity,
+            Rx,
+            Ry,
+            Rz,
+            SGate,
+            TGate,
+            Toffoli,
+            TwoBitCSwap,
+            TwoBitSwap,
+            XGate,
+            YGate,
+            ZGate,
+        )
+
+        assert GlobalPhase(exponent=1) == _map_to_bloq(
+            qml.GlobalPhase(GlobalPhase(exponent=1).exponent * np.pi, 0)
+        )
+        assert Identity() == _map_to_bloq(qml.Identity(0))
+        assert Ry(angle=np.pi / 2) == _map_to_bloq(qml.RY(np.pi / 2, 0))
+        assert Rx(angle=np.pi / 4) == _map_to_bloq(qml.RX(np.pi / 4, 0))
+        assert Rz(angle=np.pi / 3) == _map_to_bloq(qml.RZ(np.pi / 3, 0))
+        assert SGate() == _map_to_bloq(qml.S(0))
+        assert TwoBitSwap() == _map_to_bloq(qml.SWAP([0, 1]))
+        assert TwoBitCSwap() == _map_to_bloq(qml.CSWAP([0, 1, 2]))
+        assert TGate() == _map_to_bloq(qml.T(0))
+        assert XGate() == _map_to_bloq(qml.PauliX(0))
+        assert YGate() == _map_to_bloq(qml.PauliY(0))
+        assert CYGate() == _map_to_bloq(qml.CY([0, 1]))
+        assert ZGate() == _map_to_bloq(qml.PauliZ(0))
+        assert CZ() == _map_to_bloq(qml.CZ([0, 1]))
+        assert CNOT() == _map_to_bloq(qml.CNOT([0, 1]))
+        assert Toffoli() == _map_to_bloq(qml.Toffoli([0, 1, 2]))
+
+    @pytest.mark.parametrize(
+        (
+            "op",
+            "qml_call_graph",  # Computed by resources from labs or decompositions
+        ),
+        [
+            (
+                qml.QuantumPhaseEstimation(
+                    unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)
+                ),
+                # ResourceQPE
+                {
+                    (qml.Hadamard(0), True): 4,
+                    (qml.ctrl(qml.RX(0.1, wires=0), control=[1]), True): 15,
+                    (qml.adjoint(qml.QFT(wires=range(1, 5))), False): 1,
+                },
+            ),
+            (
+                qml.Superposition(
+                    coeffs=np.sqrt(np.array([1 / 3, 1 / 3, 1 / 3])),
+                    bases=np.array([[1, 1, 1], [0, 1, 0], [0, 0, 0]]),
+                    wires=[0, 1, 2],
+                    work_wire=3,
+                ),
+                # Inspired by Resource Superposition
+                {
+                    (
+                        qml.StatePrep(
+                            np.array([0.57735027, 0.57735027, 0.57735027]), wires=[2, 3], pad_with=0
+                        ),
+                        False,
+                    ): 1,
+                    (qml.CNOT([0, 1]), True): 2,
+                    (qml.MultiControlledX(wires=range(4), control_values=[1, 0, 0]), True): 4,
+                },
+            ),
+            (qml.BasisState(np.array([1, 1]), wires=[0, 1]), {(qml.X(0), True): 2}),
+            (
+                qml.QFT(wires=range(5)),
+                # From ResourceQFT
+                {
+                    (qml.H(0), True): 5,
+                    (qml.ControlledPhaseShift(1, [0, 1]), True): 10,
+                    (qml.SWAP([0, 1]), True): 2,
+                },
+            ),
+            (
+                qml.QROMStatePreparation(
+                    np.sqrt(np.array([0.5, 0.0, 0.25, 0.25])), [4, 5], [1, 2, 3], [0]
+                ),
+                {
+                    (
+                        qml.QROM(
+                            bitstrings=["001"],
+                            control_wires=[],
+                            target_wires=[1, 2, 3],
+                            work_wires=[0],
+                            clean=False,
+                        ),
+                        True,
+                    ): 1,
+                    (
+                        qml.adjoint(
+                            qml.QROM(
+                                bitstrings=["001"],
+                                control_wires=[],
+                                target_wires=[1, 2, 3],
+                                work_wires=[0],
+                                clean=False,
+                            )
+                        ),
+                        True,
+                    ): 1,
+                    (
+                        qml.QROM(
+                            bitstrings=["000", "001"],
+                            control_wires=[4],
+                            target_wires=[1, 2, 3],
+                            work_wires=[0],
+                            clean=False,
+                        ),
+                        True,
+                    ): 1,
+                    (
+                        qml.adjoint(
+                            qml.QROM(
+                                bitstrings=["000", "001"],
+                                control_wires=[4],
+                                target_wires=[1, 2, 3],
+                                work_wires=[0],
+                                clean=False,
+                            )
+                        ),
+                        True,
+                    ): 1,
+                    (qml.CRY(0.0, wires=[0, 1]), True): 6,
+                },
+            ),
+            (
+                qml.QROM(
+                    bitstrings=["000", "001"],
+                    control_wires=[4],
+                    target_wires=[1, 2, 3],
+                    work_wires=[0],
+                    clean=False,
+                ),
+                # From ResourceQROM
+                {
+                    (qml.CNOT([0, 1]), True): 1,
+                    (
+                        qml.MultiControlledX(
+                            wires=[0, 1], control_values=[True], work_wires=range(2, 3)
+                        ),
+                        True,
+                    ): 4,
+                    (qml.X(0), True): 4,
+                    (qml.CSWAP([0, 1, 2]), True): 0.0,
+                },
+            ),
+            (
+                qml.QROM(
+                    bitstrings=["001"],
+                    control_wires=[],
+                    target_wires=[1, 2, 3],
+                    work_wires=[0],
+                    clean=False,
+                ),
+                # From ResourceQROM
+                {
+                    (qml.X(0), True): 1,
+                },
+            ),
+            (
+                qml.QROM(
+                    bitstrings=["000", "001"],
+                    control_wires=[4],
+                    target_wires=[1, 2, 3],
+                    work_wires=[0],
+                    clean=True,
+                ),
+                # From ResourceQROM
+                {
+                    (qml.Hadamard(0), True): 6,
+                    (qml.CNOT([0, 1]), True): 1,
+                    (
+                        qml.MultiControlledX(
+                            wires=[0, 1], control_values=[True], work_wires=range(2, 3)
+                        ),
+                        True,
+                    ): 8,
+                    (qml.X(0), True): 8,
+                    (qml.CSWAP([0, 1, 2]), True): 0.0,
+                },
+            ),
+            (
+                qml.QROMStatePreparation(np.array([0.5, -0.5, 0.5, 0.5]), [4, 5], [1, 2, 3], [0]),
+                {
+                    (
+                        qml.QROM(
+                            bitstrings=["001"],
+                            control_wires=[],
+                            target_wires=[1, 2, 3],
+                            work_wires=[0],
+                            clean=False,
+                        ),
+                        True,
+                    ): 1,
+                    (
+                        qml.adjoint(
+                            qml.QROM(
+                                bitstrings=["001"],
+                                control_wires=[],
+                                target_wires=[1, 2, 3],
+                                work_wires=[0],
+                                clean=False,
+                            )
+                        ),
+                        True,
+                    ): 1,
+                    (
+                        qml.QROM(
+                            bitstrings=["000", "001"],
+                            control_wires=[4],
+                            target_wires=[1, 2, 3],
+                            work_wires=[0],
+                            clean=False,
+                        ),
+                        True,
+                    ): 1,
+                    (
+                        qml.adjoint(
+                            qml.QROM(
+                                bitstrings=["000", "001"],
+                                control_wires=[4],
+                                target_wires=[1, 2, 3],
+                                work_wires=[0],
+                                clean=False,
+                            )
+                        ),
+                        True,
+                    ): 1,
+                    (
+                        qml.QROM(
+                            bitstrings=["000", "000", "001", "001"],
+                            control_wires=[4, 5],
+                            target_wires=[1, 2, 3],
+                            work_wires=[0],
+                            clean=False,
+                        ),
+                        True,
+                    ): 1,
+                    (
+                        qml.adjoint(
+                            qml.QROM(
+                                bitstrings=["000", "000", "001", "001"],
+                                control_wires=[4, 5],
+                                target_wires=[1, 2, 3],
+                                work_wires=[0],
+                                clean=False,
+                            )
+                        ),
+                        True,
+                    ): 1,
+                    (qml.CRY(0.0, wires=[0, 1]), True): 6,
+                    (
+                        qml.ctrl(
+                            qml.GlobalPhase((2 * np.pi), wires=[1]),
+                            control=0,
+                        ),
+                        True,
+                    ): 3,
+                },
+            ),
+            (
+                qml.ModExp(
+                    x_wires=[0, 1],
+                    output_wires=[2, 3, 4],
+                    base=2,
+                    mod=7,
+                    work_wires=[5, 6, 7, 8, 9],
+                ),
+                {
+                    (qml.ctrl(qml.adjoint(qml.QFT(range(4))), control=[4]), False): 1,
+                    (qml.ctrl(qml.QFT(range(4)), control=[4]), False): 1,
+                    (qml.Toffoli([0, 1, 2]), True): 6,
+                },
+            ),
+            (
+                qml.ModExp(
+                    x_wires=[0, 1, 2],
+                    output_wires=[3, 4, 5],
+                    base=3,
+                    mod=8,
+                    work_wires=[6, 7, 8, 9, 10],
+                ),
+                {
+                    (qml.ctrl(qml.QFT(range(3)), control=[4]), False): 1,
+                    (qml.ctrl(qml.adjoint(qml.QFT(range(3))), control=[4]), False): 1,
+                    (qml.Toffoli([0, 1, 2]), True): 21,
+                },
+            ),
+            (
+                qml.QSVT(
+                    UA=qml.H(0),
+                    projectors=[qml.RZ(-2 * theta, wires=0) for theta in (1.23, -0.5, -0.3)],
+                ),
+                {
+                    (qml.RZ(phi=-2.46, wires=0), True): 1,
+                    (qml.RZ(phi=1.0, wires=0), True): 1,
+                    (qml.Hadamard(0), True): 2,
+                    (qml.RZ(phi=0.6, wires=0), True): 1,
+                },
+            ),
+            (
+                qml.TrotterizedQfunc(
+                    0.1,
+                    *(0.12, -3.45),
+                    qfunc=lambda time, theta, phi, wires, flip: (
+                        qml.RX(time * theta, wires[0]),
+                        qml.RY(time * phi, wires[1]),
+                        qml.CNOT(wires=wires[:2]) if flip else None,
+                    ),
+                    n=1,
+                    order=2,
+                    wires=["a", "b"],
+                    flip=True,
+                ),
+                {
+                    (qml.RX(phi=0.012, wires=[0]), True): 2,
+                    (qml.RY(phi=-0.34500000000000003, wires=[0]), True): 2,
+                    (qml.CNOT(wires=[0, 1]), True): 2,
+                },
+            ),
+            (
+                qml.TrotterizedQfunc(
+                    0.1,
+                    *(0.12, -3.45),
+                    qfunc=lambda time, theta, phi, wires, flip: (
+                        qml.RX(time * theta, wires[0]),
+                        qml.RY(time * phi, wires[1]),
+                        qml.CNOT(wires=wires[:2]) if flip else None,
+                    ),
+                    n=1,
+                    order=1,
+                    wires=["a", "b"],
+                    flip=True,
+                ),
+                {
+                    (qml.RX(phi=0.012, wires=[0]), True): 1,
+                    (qml.RY(phi=-0.34500000000000003, wires=[0]), True): 1,
+                    (qml.CNOT(wires=[0, 1]), True): 1,
+                },
+            ),
+            (
+                qml.Select(ops=[qml.X(2), qml.QFT(wires=[2, 3, 4])], control=[0, 1]),
+                {
+                    (qml.X(wires=[2]), True): 2,
+                    (qml.ctrl(qml.X(2), control=[0]), True): 1,
+                    (qml.ctrl(qml.QFT(wires=[2, 3, 4]), control=[0]), True): 1,
+                },
+            ),
+            (
+                qml.StatePrep(state=[0.5, 0.5, 0.5, 0.5, 0.25, 0.25, 0.25, 0.25], wires=range(3)),
+                {(qml.RZ(0, wires=[0]), True): 27, (qml.CNOT([0, 1]), True): 16},
+            ),
+        ],
+    )
+    def test_build_call_graph(self, op, qml_call_graph):
+        """ "Tests that the defined call_graphs match the expected decompostions"""
+        bloq_call_graph = {}
+
+        for k, v in qml_call_graph.items():  # k is a tuple of (op, bool)
+            bloq_call_graph[qml.to_bloq(k[0], map_ops=k[1])] = v
+
+        call_graph = _get_op_call_graph(op)
+        assert dict(call_graph) == bloq_call_graph
+
+    @pytest.mark.parametrize(
+        (
+            "op",
+            "qt_bloq",
+        ),
+        [
+            (
+                qml.QuantumPhaseEstimation(
+                    unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)
+                ),
+                "qpe_bloq",
+            ),
+            (qml.QFT(wires=range(4)), "qft_bloq"),
+            (
+                qml.ModExp(
+                    x_wires=[0, 1],
+                    output_wires=[2, 3, 4],
+                    base=2,
+                    mod=7,
+                    work_wires=[5, 6, 7, 8, 9],
+                ),
+                "modexp_bloq",
+            ),
+            (
+                qml.QROM(
+                    bitstrings=["010", "111", "110", "000"],
+                    control_wires=[0, 1],
+                    target_wires=[2, 3, 4],
+                    work_wires=[5, 6, 7],
+                ),
+                "qrom_bloq_clean",
+            ),
+            (
+                qml.QROM(
+                    bitstrings=["010", "111", "110", "000"],
+                    control_wires=[0, 1],
+                    target_wires=[2, 3, 4],
+                    work_wires=[5, 6, 7],
+                    clean=False,
+                ),
+                "qrom_bloq_dirty",
+            ),
+        ],
+    )
+    def test_default_mapping(self, op, qt_bloq):
+        """Tests that the defined default maps match the expected qualtran bloq"""
+
+        def _build_expected_qualtran_bloq(qt_bloq):
+            """Factory function inside for parametrization of test cases"""
+            from qualtran.bloqs.cryptography.rsa import ModExp
+            from qualtran.bloqs.data_loading.qroam_clean import QROAMClean
+            from qualtran.bloqs.data_loading.select_swap_qrom import SelectSwapQROM
+            from qualtran.bloqs.phase_estimation import RectangularWindowState
+            from qualtran.bloqs.phase_estimation.text_book_qpe import TextbookQPE
+            from qualtran.bloqs.qft import QFTTextBook
+
+            qualtran_bloqs = {
+                "qpe_bloq": TextbookQPE(
+                    unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                    ctrl_state_prep=RectangularWindowState(4),
+                ),
+                "qft_bloq": QFTTextBook(4),
+                "modexp_bloq": ModExp(base=2, mod=7, exp_bitsize=2, x_bitsize=3),
+                "qrom_bloq_clean": QROAMClean.build_from_data([2, 7, 6, 0]),
+                "qrom_bloq_dirty": SelectSwapQROM.build_from_data([2, 7, 6, 0]),
+            }
+
+            return qualtran_bloqs[qt_bloq]
+
+        qt_qpe = qml.to_bloq(op, map_ops=True)
+        assert qt_qpe == _build_expected_qualtran_bloq(qt_bloq)
+
+    @pytest.mark.parametrize(
+        (
+            "op",
+            "custom_map",
+            "qt_bloq",
+        ),
+        [
+            (
+                qml.QuantumPhaseEstimation(
+                    unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)
+                ),
+                "qpe_custom_mapping",
+                "qpe_custom_bloq",
+            ),
+            # Tests the behaviour of using custom mapping for ops without default mappings
+            (
+                qml.QSVT(qml.H(0), [qml.RZ(-2 * theta, wires=0) for theta in (1.23, -0.5, 4)]),
+                "qsvt_custom_mapping",
+                "qsvt_custom_bloq",
+            ),
+            (qml.QFT(wires=range(4)), "qft_custom_mapping", "qft_custom_bloq"),
+            (
+                qml.ModExp(
+                    x_wires=[0, 1],
+                    output_wires=[2, 3, 4],
+                    base=2,
+                    mod=7,
+                    work_wires=[5, 6, 7, 8, 9],
+                ),
+                "modexp_custom_mapping",
+                "modexp_custom_bloq",
+            ),
+            (
+                qml.QROM(
+                    bitstrings=["010", "111", "110", "000"],
+                    control_wires=[0, 1],
+                    target_wires=[2, 3, 4],
+                    work_wires=[5, 6, 7],
+                ),
+                "qrom_custom_mapping",
+                "qrom_custom_bloq",
+            ),
+        ],
+    )
+    def test_custom_mapping(self, op, custom_map, qt_bloq):
+        """Tests that custom mapping maps the expected qualtran bloq"""
+
+        def _build_expected_qualtran_bloq(qt_bloq):
+            """Factory function to build expected Qualtran bloq inside for parametrization of test cases"""
+            from qualtran.bloqs.phase_estimation import LPResourceState
+            from qualtran.bloqs.phase_estimation.text_book_qpe import TextbookQPE
+
+            qualtran_bloqs = {
+                "qpe_custom_bloq": TextbookQPE(
+                    unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                    ctrl_state_prep=LPResourceState(4),
+                ),
+                "qsvt_custom_bloq": TextbookQPE(
+                    unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                    ctrl_state_prep=LPResourceState(4),
+                ),
+                "qft_custom_bloq": TextbookQPE(
+                    unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                    ctrl_state_prep=LPResourceState(4),
+                ),
+                "modexp_custom_bloq": TextbookQPE(
+                    unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                    ctrl_state_prep=LPResourceState(4),
+                ),
+                "qrom_custom_bloq": TextbookQPE(
+                    unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                    ctrl_state_prep=LPResourceState(4),
+                ),
+            }
+
+            return qualtran_bloqs[qt_bloq]
+
+        def _build_custom_map(custom_map):
+            """Factory function to build custom maps for parametrization of test cases"""
+            from qualtran.bloqs.phase_estimation import LPResourceState
+            from qualtran.bloqs.phase_estimation.text_book_qpe import TextbookQPE
+
+            custom_mapping = {
+                "qpe_custom_mapping": {
+                    qml.QuantumPhaseEstimation(
+                        unitary=qml.RX(0.1, wires=0), estimation_wires=range(1, 5)
+                    ): TextbookQPE(
+                        unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                        ctrl_state_prep=LPResourceState(4),
+                    )
+                },
+                "qsvt_custom_mapping": {
+                    qml.QSVT(
+                        qml.H(0), [qml.RZ(-2 * theta, wires=0) for theta in (1.23, -0.5, 4)]
+                    ): TextbookQPE(
+                        unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                        ctrl_state_prep=LPResourceState(4),
+                    )
+                },
+                "qft_custom_mapping": {
+                    qml.QFT(wires=range(4)): TextbookQPE(
+                        unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                        ctrl_state_prep=LPResourceState(4),
+                    )
+                },
+                "modexp_custom_mapping": {
+                    qml.ModExp(
+                        x_wires=[0, 1],
+                        output_wires=[2, 3, 4],
+                        base=2,
+                        mod=7,
+                        work_wires=[5, 6, 7, 8, 9],
+                    ): TextbookQPE(
+                        unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                        ctrl_state_prep=LPResourceState(4),
+                    )
+                },
+                "qrom_custom_mapping": {
+                    qml.QROM(
+                        bitstrings=["010", "111", "110", "000"],
+                        control_wires=[0, 1],
+                        target_wires=[2, 3, 4],
+                        work_wires=[5, 6, 7],
+                    ): TextbookQPE(
+                        unitary=qml.to_bloq(qml.RX(0.1, wires=0)),
+                        ctrl_state_prep=LPResourceState(4),
+                    )
+                },
+            }
+
+            return custom_mapping[custom_map]
+
+        qt_qpe = qml.to_bloq(op, map_ops=True, custom_mapping=_build_custom_map(custom_map))
+        assert qt_qpe == _build_expected_qualtran_bloq(qt_bloq)
+
+    # pylint: disable=redefined-outer-name, protected-access
+    def test_initialization_with_tuple(self, qubits, dtypes):
+        """Tests standard initialization with a tuple of qubits."""
+        q0, q1 = qubits
+        _, dtype_uint = dtypes
+        qubits_tuple = (q0, q1)
+
+        qreg = _QReg(qubits=qubits_tuple, dtype=dtype_uint)
+
+        assert isinstance(qreg.qubits, tuple)
+
+        assert qreg.qubits == qubits_tuple
+        assert qreg.dtype == dtype_uint
+        assert qreg._initialized is True
+
+    # pylint: disable=redefined-outer-name
+    def test_initialization_with_single_qubit(self, qubits, dtypes):
+        """Tests that a single qubit is correctly wrapped in a tuple."""
+        q0, _ = qubits
+        dtype_bit, _ = dtypes
+
+        qreg = _QReg(qubits=q0, dtype=dtype_bit)
+
+        assert qreg.__repr__() == "_QReg(qubits=(cirq.LineQubit(0),), dtype=QBit())"
+        assert isinstance(qreg.qubits, tuple)
+        assert qreg.qubits == (q0,)
+        assert qreg.dtype == dtype_bit
+
+    # pylint: disable=redefined-outer-name
+    def test_immutability_raises_error_on_attribute_change(self, qubits, dtypes):
+        """Tests that changing an attribute after initialization raises an AttributeError."""
+        q0, q1 = qubits
+        dtype_bit, dtype_uint = dtypes
+        qreg = _QReg(qubits=(q0,), dtype=dtype_bit)
+
+        with pytest.raises(AttributeError, match="Cannot set attribute 'qubits'"):
+            qreg.qubits = (q1,)
+
+        with pytest.raises(AttributeError, match="Cannot set attribute 'dtype'"):
+            qreg.dtype = dtype_uint
+
+        with pytest.raises(AttributeError, match="Cannot set attribute 'new_attr'"):
+            qreg.new_attr = "some_value"
+
+    # pylint: disable=redefined-outer-name
+    def test_equality_ignores_dtype(self, qubits, dtypes):
+        """Tests the core feature: equality should only depend on qubits, not dtype."""
+        q0, _ = qubits
+        dtype_bit, dtype_uint = dtypes
+
+        qreg1 = _QReg(qubits=(q0,), dtype=dtype_bit)
+        qreg2 = _QReg(qubits=(q0,), dtype=dtype_uint)
+
+        assert qreg1 == qreg2
+
+    # pylint: disable=redefined-outer-name
+    def test_hash_ignores_dtype(self, qubits, dtypes):
+        """Tests that the hash also only depends on the qubits."""
+        q0, _ = qubits
+        dtype_bit, dtype_uint = dtypes
+
+        qreg1 = _QReg(qubits=(q0,), dtype=dtype_bit)
+        qreg2 = _QReg(qubits=(q0,), dtype=dtype_uint)
+
+        assert hash(qreg1) == hash(qreg2)
+
+    # pylint: disable=redefined-outer-name
+    def test_inequality_for_different_qubits(self, qubits, dtypes):
+        """Tests that instances with different qubits are not equal."""
+        q0, q1 = qubits
+        dtype_bit, _ = dtypes
+
+        qreg1 = _QReg(qubits=(q0,), dtype=dtype_bit)
+        qreg2 = _QReg(qubits=(q1,), dtype=dtype_bit)
+        qreg3 = _QReg(qubits=(q0, q1), dtype=dtype_bit)
+
+        assert qreg1 != qreg2
+        assert qreg1 != qreg3
+
+    # pylint: disable=redefined-outer-name
+    def test_inequality_for_different_types(self, qubits, dtypes):
+        """Tests that comparison with other types returns NotImplemented/False."""
+        q0, _ = qubits
+        dtype_bit, _ = dtypes
+        qreg = _QReg(qubits=(q0,), dtype=dtype_bit)
+
+        assert qreg != "not_a_qreg"
+        assert qreg != (q0,)
+        assert qreg is not None