API Reference

Auto-generated from the library's docstrings.

Core

hdh.hdh.HDH

A Hybrid Dependency Hypergraph: a quantum workload as states (nodes) joined by the operations between them (hyperedges).

Usually built by a model (Circuit, MBQC, QW, QCA) or a converter such as from_qiskit. Each node is the state of one wire at one timestep, with ID "<wire>_t<time>". Core attributes have a readable name and a formal one matching the HDH paper's notation; both refer to the same object.

Examples:

>>> hdh = HDH()
>>> a = hdh.add_node("q0", 0)
>>> b = hdh.add_node("q0", 1)
>>> edge = hdh.add_hyperedge({a, b}, "q", name="h")
>>> sorted(hdh.nodes)
['q0_t0', 'q0_t1']
>>> hdh.hyperedge_types[edge], hdh.gate_name[edge]
('q', 'h')
Attributes:
  • nodes (S) –

    All node IDs.

  • hyperedges (C) –

    All hyperedges, each a frozenset of node IDs.

  • timesteps (T) –

    All timesteps in use.

  • node_types (sigma) –

    Node -> "q" or "c".

  • hyperedge_types (tau) –

    Hyperedge -> "q" or "c".

  • node_realisation (upsilon) –

    Node -> "a" (always happens) or "p" (only if a classical condition holds).

  • hyperedge_realisation (phi) –

    Hyperedge -> "a" or "p".

  • time_map (Dict[NodeID, TimeStep]) –

    Node -> its timestep.

  • wire_of (Dict[NodeID, str]) –

    Node -> the wire it is a state of. Partitioners count capacity per quantum wire.

  • gate_name (Dict[Hyperedge, str]) –

    Hyperedge -> the operation that produced it, e.g. "cx_stage2".

  • gate_params (Dict[Hyperedge, List[float]]) –

    Hyperedge -> gate parameters, when recorded.

  • edge_args (Dict[Hyperedge, Tuple[List[int], List[int], List[bool]]]) –

    Hyperedge -> data the converters use to rebuild a circuit.

  • edge_role (Dict[Hyperedge, str]) –

    Hyperedge -> "teledata" or "telegate", once assigned.

  • edge_metadata (Dict[Hyperedge, Dict]) –

    Free-form per-hyperedge metadata.

  • motifs –

    Reserved; unused by the core API.

Source code in hdh\hdh.py
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
class HDH:
    """A Hybrid Dependency Hypergraph: a quantum workload as states (nodes)
    joined by the operations between them (hyperedges).

    Usually built by a model (`Circuit`, `MBQC`, `QW`, `QCA`) or a
    converter such as `from_qiskit`. Each node is the state of one wire at one
    timestep, with ID ``"<wire>_t<time>"``. Core attributes have a readable
    name and a formal one matching the HDH paper's notation; both refer to the
    same object.

    Examples:
        >>> hdh = HDH()
        >>> a = hdh.add_node("q0", 0)
        >>> b = hdh.add_node("q0", 1)
        >>> edge = hdh.add_hyperedge({a, b}, "q", name="h")
        >>> sorted(hdh.nodes)
        ['q0_t0', 'q0_t1']
        >>> hdh.hyperedge_types[edge], hdh.gate_name[edge]
        ('q', 'h')

    Attributes:
        nodes (S): All node IDs.
        hyperedges (C): All hyperedges, each a frozenset of node IDs.
        timesteps (T): All timesteps in use.
        node_types (sigma): Node -> ``"q"`` or ``"c"``.
        hyperedge_types (tau): Hyperedge -> ``"q"`` or ``"c"``.
        node_realisation (upsilon): Node -> ``"a"`` (always happens) or
            ``"p"`` (only if a classical condition holds).
        hyperedge_realisation (phi): Hyperedge -> ``"a"`` or ``"p"``.
        time_map: Node -> its timestep.
        wire_of: Node -> the wire it is a state of. Partitioners count
            capacity per quantum wire.
        gate_name: Hyperedge -> the operation that produced it, e.g.
            ``"cx_stage2"``.
        gate_params: Hyperedge -> gate parameters, when recorded.
        edge_args: Hyperedge -> data the converters use to rebuild a circuit.
        edge_role: Hyperedge -> ``"teledata"`` or ``"telegate"``, once assigned.
        edge_metadata: Free-form per-hyperedge metadata.
        motifs: Reserved; unused by the core API.
    """

    def __init__(self):
        self.S: Set[NodeID] = set()
        self.C: Set[Hyperedge] = set()
        self.T: Set[TimeStep] = set()
        self.sigma: Dict[NodeID, str] = {}  # node types, a NodeType value
        self.tau: Dict[Hyperedge, str] = {}  # hyperedge types, a NodeType value
        self.upsilon: Dict[NodeID, str] = {} # node realization, a Realisation value
        self.phi: Dict[Hyperedge, str] = {} # hyperedge realization, a Realisation value
        self.time_map: Dict[NodeID, TimeStep] = {}  # f: S -> T
        self.wire_of: Dict[NodeID, str] = {}  # node -> the wire it is a state of
        self.gate_name: Dict[Hyperedge, str] = {}  # maps hyperedge → gate name string
        self.gate_params: Dict[Hyperedge, List[float]] = {}  # maps hyperedge → rotation params, if any
        self.edge_args: Dict[Hyperedge, Tuple[List[int], List[int], List[bool]]] = {} #mapping for nackwards translations
        self.edge_role: Dict[Hyperedge, str] = {}  # an EdgeRole value -> for primitive implementation
        self.motifs = {}
        self.edge_metadata: Dict[Hyperedge, Dict] = {}

    # Readable names for the formal attributes above (same objects, not copies).

    @property
    def nodes(self) -> Set[NodeID]:
        """All node IDs (`S` in the paper's notation)."""
        return self.S

    @property
    def hyperedges(self) -> Set[Hyperedge]:
        """All hyperedges (`C`)."""
        return self.C

    @property
    def timesteps(self) -> Set[TimeStep]:
        """All distinct timesteps (`T`)."""
        return self.T

    @property
    def node_types(self) -> Dict[NodeID, str]:
        """Node ID -> `"q"` or `"c"` (`sigma`)."""
        return self.sigma

    @property
    def hyperedge_types(self) -> Dict[Hyperedge, str]:
        """Hyperedge -> `"q"` or `"c"` (`tau`)."""
        return self.tau

    @property
    def node_realisation(self) -> Dict[NodeID, str]:
        """Node ID -> `"a"` or `"p"` (`upsilon`)."""
        return self.upsilon

    @property
    def hyperedge_realisation(self) -> Dict[Hyperedge, str]:
        """Hyperedge -> `"a"` or `"p"` (`phi`)."""
        return self.phi

    @staticmethod
    def node_id(wire: str, time: TimeStep) -> NodeID:
        """The ID `add_node` gives the state of `wire` at `time`: ``"<wire>_t<time>"``."""
        return f"{wire}_t{time}"

    def add_node(self, wire: str, time: TimeStep, node_type: NodeType = NodeType.QUANTUM,
                 node_real: NodeReal = Realisation.ACTUAL) -> NodeID:
        """Add the state of `wire` at `time` and return its node ID.

        The ID is ``"<wire>_t<time>"`` (see `node_id`) and the wire is recorded
        in `wire_of`. Re-adding an existing state is a no-op apart from updating
        `node_real`.

        Args:
            wire: Qubit, bit or model label, e.g. ``"q0"``, ``"c1"`` or ``"a"``.
            time: Timestep of this state.
            node_type: ``"q"`` (default) or ``"c"``, or a `NodeType`.
            node_real: ``"a"`` (default) or ``"p"``, or a `Realisation`.

        Returns:
            The node ID.

        Raises:
            TypeError: If called with the pre-0.5 signature
                ``add_node(node_id, node_type, time)``.
            ValueError: On an invalid type, or if the state already exists with
                another type.

        Examples:
            >>> hdh = HDH()
            >>> hdh.add_node("q0", 1)
            'q0_t1'
            >>> hdh.add_node("c0", 2, "c")
            'c0_t2'
        """
        if not isinstance(time, numbers.Integral) or isinstance(time, bool):
            raise TypeError(
                f"add_node(wire, time, node_type) expects an integer time, got {time!r}. "
                f"Since hdh 0.5 the node ID is built from the wire and time: "
                f"write add_node('q0', 1, 'q') instead of add_node('q0_t1', 'q', 1)."
            )
        if not isinstance(wire, str) or not wire:
            raise TypeError(f"wire must be a non-empty string, got {wire!r}")
        time = int(time)
        node_type = _coerce(NodeType, node_type, "node type")
        node_real = _coerce(Realisation, node_real, "node realisation")
        node_id = self.node_id(wire, time)
        existing_type = self.sigma.get(node_id)
        if existing_type is not None and existing_type != node_type:
            raise ValueError(
                f"Node '{node_id}' already exists with type '{existing_type}'; "
                f"cannot redefine it as type '{node_type}'. This usually means two "
                f"different logical values were mapped to the same wire label."
            )
        self.S.add(node_id)
        self.sigma[node_id] = node_type
        self.wire_of[node_id] = wire
        self.time_map[node_id] = time
        self.T.add(time)
        self.upsilon[node_id] = node_real
        return node_id

    def add_hyperedge(self, node_ids: Set[NodeID], edge_type: EdgeType, name: Optional[str] = None, node_real: EdgeReal = "a", role: Optional[EdgeRole] = None) -> Hyperedge:
        """Connect `node_ids` with a hyperedge representing one operation.

        Args:
            node_ids: The states the operation touches, inputs and outputs together.
            edge_type: ``"q"`` or ``"c"``, or a `NodeType`.
            name: Operation name, stored lower-cased in `gate_name`.
            node_real: ``"a"`` (default) or ``"p"``, or a `Realisation`.
            role: ``"teledata"`` or ``"telegate"``, usually assigned later by a
                partitioning pass.

        Returns:
            The hyperedge (a frozenset), used as the key into `hyperedge_types`,
            `gate_name` and the other per-edge maps.

        Raises:
            ValueError: If `edge_type`, `node_real` or `role` is invalid.

        Examples:
            >>> hdh = HDH()
            >>> q, c = hdh.add_node("q0", 0), hdh.add_node("c0", 1, "c")
            >>> edge = hdh.add_hyperedge({q, c}, "c", name="measure")
            >>> sorted(edge)
            ['c0_t1', 'q0_t0']
        """
        edge_type = _coerce(NodeType, edge_type, "edge type")
        node_real = _coerce(Realisation, node_real, "edge realisation")
        if role:
            role = _coerce(EdgeRole, role, "edge role")
        edge = frozenset(node_ids)
        self.C.add(edge)
        self.tau[edge] = edge_type
        self.phi[edge] = node_real
        if name:
            self.gate_name[edge] = name.lower()
        if role:
            self.edge_role[edge] = role
        return edge

    def get_ancestry(self, node: NodeID) -> Set[NodeID]:
        """Return nodes with paths ending at `node` and earlier time steps."""
        return {
            s for s in self.S
            if self.time_map[s] <= self.time_map[node] and self._path_exists(s, node)
        }

    def get_lineage(self, node: NodeID) -> Set[NodeID]:
        """Return nodes reachable from `node` with later time steps."""
        return {
            s for s in self.S
            if self.time_map[s] >= self.time_map[node] and self._path_exists(node, s)
        }

    def _path_exists(self, start: NodeID, end: NodeID) -> bool:
        """DFS to find a time-respecting path from `start` to `end`."""
        visited = set()
        stack = [start]
        while stack:
            current = stack.pop()
            if current == end:
                return True
            visited.add(current)
            neighbors = {
                neighbor
                for edge in self.C if current in edge
                for neighbor in edge
                if neighbor != current and self.time_map[neighbor] > self.time_map[current]
            }
            stack.extend(neighbors - visited)
        return False

    def get_num_qubits(self) -> int:
        """Return the number of logical qubits (quantum wires).

        Circuit-style wires ``q<index>`` report `max(index) + 1`, so a circuit
        using only qubits 0 and 4 reports 5, matching the register size a
        converter needs. If any quantum wire is named otherwise (MBQC or QCA
        labels), the count of distinct quantum wires is returned instead.

        Returns:
            int: number of qubits, or 0 if there are no quantum nodes.
        """
        wires = {self.wire_of[n] for n in self.S if self.sigma[n] == NodeType.QUANTUM}
        indices = [int(w[1:]) for w in wires if w[:1] == "q" and w[1:].isdigit()]
        if len(indices) != len(wires):
            return len(wires)
        return max(indices) + 1 if indices else 0

hyperedge_realisation property

Hyperedge -> "a" or "p" (phi).

hyperedge_types property

Hyperedge -> "q" or "c" (tau).

hyperedges property

All hyperedges (C).

node_realisation property

Node ID -> "a" or "p" (upsilon).

node_types property

Node ID -> "q" or "c" (sigma).

nodes property

All node IDs (S in the paper's notation).

timesteps property

All distinct timesteps (T).

add_hyperedge(node_ids, edge_type, name=None, node_real='a', role=None)

Connect node_ids with a hyperedge representing one operation.

Parameters:
  • node_ids (Set[NodeID]) –

    The states the operation touches, inputs and outputs together.

  • edge_type (EdgeType) –

    "q" or "c", or a NodeType.

  • name (Optional[str], default: None ) –

    Operation name, stored lower-cased in gate_name.

  • node_real (EdgeReal, default: 'a' ) –

    "a" (default) or "p", or a Realisation.

  • role (Optional[EdgeRole], default: None ) –

    "teledata" or "telegate", usually assigned later by a partitioning pass.

Returns:
  • Hyperedge –

    The hyperedge (a frozenset), used as the key into hyperedge_types,

  • Hyperedge –

    gate_name and the other per-edge maps.

Raises:
  • ValueError –

    If edge_type, node_real or role is invalid.

Examples:

>>> hdh = HDH()
>>> q, c = hdh.add_node("q0", 0), hdh.add_node("c0", 1, "c")
>>> edge = hdh.add_hyperedge({q, c}, "c", name="measure")
>>> sorted(edge)
['c0_t1', 'q0_t0']
Source code in hdh\hdh.py
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def add_hyperedge(self, node_ids: Set[NodeID], edge_type: EdgeType, name: Optional[str] = None, node_real: EdgeReal = "a", role: Optional[EdgeRole] = None) -> Hyperedge:
    """Connect `node_ids` with a hyperedge representing one operation.

    Args:
        node_ids: The states the operation touches, inputs and outputs together.
        edge_type: ``"q"`` or ``"c"``, or a `NodeType`.
        name: Operation name, stored lower-cased in `gate_name`.
        node_real: ``"a"`` (default) or ``"p"``, or a `Realisation`.
        role: ``"teledata"`` or ``"telegate"``, usually assigned later by a
            partitioning pass.

    Returns:
        The hyperedge (a frozenset), used as the key into `hyperedge_types`,
        `gate_name` and the other per-edge maps.

    Raises:
        ValueError: If `edge_type`, `node_real` or `role` is invalid.

    Examples:
        >>> hdh = HDH()
        >>> q, c = hdh.add_node("q0", 0), hdh.add_node("c0", 1, "c")
        >>> edge = hdh.add_hyperedge({q, c}, "c", name="measure")
        >>> sorted(edge)
        ['c0_t1', 'q0_t0']
    """
    edge_type = _coerce(NodeType, edge_type, "edge type")
    node_real = _coerce(Realisation, node_real, "edge realisation")
    if role:
        role = _coerce(EdgeRole, role, "edge role")
    edge = frozenset(node_ids)
    self.C.add(edge)
    self.tau[edge] = edge_type
    self.phi[edge] = node_real
    if name:
        self.gate_name[edge] = name.lower()
    if role:
        self.edge_role[edge] = role
    return edge

add_node(wire, time, node_type=NodeType.QUANTUM, node_real=Realisation.ACTUAL)

Add the state of wire at time and return its node ID.

The ID is "<wire>_t<time>" (see node_id) and the wire is recorded in wire_of. Re-adding an existing state is a no-op apart from updating node_real.

Parameters:
  • wire (str) –

    Qubit, bit or model label, e.g. "q0", "c1" or "a".

  • time (TimeStep) –

    Timestep of this state.

  • node_type (NodeType, default: QUANTUM ) –

    "q" (default) or "c", or a NodeType.

  • node_real (NodeReal, default: ACTUAL ) –

    "a" (default) or "p", or a Realisation.

Returns:
  • NodeID –

    The node ID.

Raises:
  • TypeError –

    If called with the pre-0.5 signature add_node(node_id, node_type, time).

  • ValueError –

    On an invalid type, or if the state already exists with another type.

Examples:

>>> hdh = HDH()
>>> hdh.add_node("q0", 1)
'q0_t1'
>>> hdh.add_node("c0", 2, "c")
'c0_t2'
Source code in hdh\hdh.py
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def add_node(self, wire: str, time: TimeStep, node_type: NodeType = NodeType.QUANTUM,
             node_real: NodeReal = Realisation.ACTUAL) -> NodeID:
    """Add the state of `wire` at `time` and return its node ID.

    The ID is ``"<wire>_t<time>"`` (see `node_id`) and the wire is recorded
    in `wire_of`. Re-adding an existing state is a no-op apart from updating
    `node_real`.

    Args:
        wire: Qubit, bit or model label, e.g. ``"q0"``, ``"c1"`` or ``"a"``.
        time: Timestep of this state.
        node_type: ``"q"`` (default) or ``"c"``, or a `NodeType`.
        node_real: ``"a"`` (default) or ``"p"``, or a `Realisation`.

    Returns:
        The node ID.

    Raises:
        TypeError: If called with the pre-0.5 signature
            ``add_node(node_id, node_type, time)``.
        ValueError: On an invalid type, or if the state already exists with
            another type.

    Examples:
        >>> hdh = HDH()
        >>> hdh.add_node("q0", 1)
        'q0_t1'
        >>> hdh.add_node("c0", 2, "c")
        'c0_t2'
    """
    if not isinstance(time, numbers.Integral) or isinstance(time, bool):
        raise TypeError(
            f"add_node(wire, time, node_type) expects an integer time, got {time!r}. "
            f"Since hdh 0.5 the node ID is built from the wire and time: "
            f"write add_node('q0', 1, 'q') instead of add_node('q0_t1', 'q', 1)."
        )
    if not isinstance(wire, str) or not wire:
        raise TypeError(f"wire must be a non-empty string, got {wire!r}")
    time = int(time)
    node_type = _coerce(NodeType, node_type, "node type")
    node_real = _coerce(Realisation, node_real, "node realisation")
    node_id = self.node_id(wire, time)
    existing_type = self.sigma.get(node_id)
    if existing_type is not None and existing_type != node_type:
        raise ValueError(
            f"Node '{node_id}' already exists with type '{existing_type}'; "
            f"cannot redefine it as type '{node_type}'. This usually means two "
            f"different logical values were mapped to the same wire label."
        )
    self.S.add(node_id)
    self.sigma[node_id] = node_type
    self.wire_of[node_id] = wire
    self.time_map[node_id] = time
    self.T.add(time)
    self.upsilon[node_id] = node_real
    return node_id

get_ancestry(node)

Return nodes with paths ending at node and earlier time steps.

Source code in hdh\hdh.py
253
254
255
256
257
258
def get_ancestry(self, node: NodeID) -> Set[NodeID]:
    """Return nodes with paths ending at `node` and earlier time steps."""
    return {
        s for s in self.S
        if self.time_map[s] <= self.time_map[node] and self._path_exists(s, node)
    }

get_lineage(node)

Return nodes reachable from node with later time steps.

Source code in hdh\hdh.py
260
261
262
263
264
265
def get_lineage(self, node: NodeID) -> Set[NodeID]:
    """Return nodes reachable from `node` with later time steps."""
    return {
        s for s in self.S
        if self.time_map[s] >= self.time_map[node] and self._path_exists(node, s)
    }

get_num_qubits()

Return the number of logical qubits (quantum wires).

Circuit-style wires q<index> report max(index) + 1, so a circuit using only qubits 0 and 4 reports 5, matching the register size a converter needs. If any quantum wire is named otherwise (MBQC or QCA labels), the count of distinct quantum wires is returned instead.

Returns:
  • int( int ) –

    number of qubits, or 0 if there are no quantum nodes.

Source code in hdh\hdh.py
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def get_num_qubits(self) -> int:
    """Return the number of logical qubits (quantum wires).

    Circuit-style wires ``q<index>`` report `max(index) + 1`, so a circuit
    using only qubits 0 and 4 reports 5, matching the register size a
    converter needs. If any quantum wire is named otherwise (MBQC or QCA
    labels), the count of distinct quantum wires is returned instead.

    Returns:
        int: number of qubits, or 0 if there are no quantum nodes.
    """
    wires = {self.wire_of[n] for n in self.S if self.sigma[n] == NodeType.QUANTUM}
    indices = [int(w[1:]) for w in wires if w[:1] == "q" and w[1:].isdigit()]
    if len(indices) != len(wires):
        return len(wires)
    return max(indices) + 1 if indices else 0

node_id(wire, time) staticmethod

The ID add_node gives the state of wire at time: "<wire>_t<time>".

Source code in hdh\hdh.py
152
153
154
155
@staticmethod
def node_id(wire: str, time: TimeStep) -> NodeID:
    """The ID `add_node` gives the state of `wire` at `time`: ``"<wire>_t<time>"``."""
    return f"{wire}_t{time}"

hdh.hdh.NodeType

Bases: _StrEnum

Whether a node (or hyperedge) carries quantum or classical information.

Members compare equal to their values, so "q" and NodeType.QUANTUM are interchangeable wherever a type is expected.

Source code in hdh\hdh.py
14
15
16
17
18
19
20
21
class NodeType(_StrEnum):
    """Whether a node (or hyperedge) carries quantum or classical information.

    Members compare equal to their values, so ``"q"`` and
    ``NodeType.QUANTUM`` are interchangeable wherever a type is expected.
    """
    QUANTUM = "q"
    CLASSICAL = "c"

hdh.hdh.Realisation

Bases: _StrEnum

Whether a node (or hyperedge) always happens, or only if a classical condition holds.

Source code in hdh\hdh.py
24
25
26
27
28
class Realisation(_StrEnum):
    """Whether a node (or hyperedge) always happens, or only if a classical
    condition holds."""
    ACTUAL = "a"
    PREDICTED = "p"

hdh.hdh.EdgeRole

Bases: _StrEnum

Distribution primitive assigned to a cut hyperedge.

Source code in hdh\hdh.py
31
32
33
34
class EdgeRole(_StrEnum):
    """Distribution primitive assigned to a cut hyperedge."""
    TELEDATA = "teledata"
    TELEGATE = "telegate"

Models

hdh.models.base.Model

Bases: Protocol

What a computational model must provide to be used with HDH.

A model is any object that can translate the computation it describes into an HDH. Circuit, MBQC, QW and QCA all satisfy this Protocol; a new model only needs a build_hdh method, no base class. tests/test_model_protocol.py checks every registered model against the invariants below, so a new model should be added to its registry.

The HDH returned by build_hdh must:

  • create every node with HDH.add_node, so each has a type, a time and a wire (qubit, bit, or model label);
  • keep each wire a single type (a label used for a classical result is never reused as a quantum state);
  • only connect nodes that exist;
  • keep quantum hyperedges free of classical nodes.

Examples:

>>> from hdh.models.circuit import Circuit
>>> isinstance(Circuit(), Model)
True
Source code in hdh\models\base.py
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@runtime_checkable
class Model(Protocol):
    """What a computational model must provide to be used with HDH.

    A model is any object that can translate the computation it describes
    into an `HDH`. `Circuit`, `MBQC`, `QW` and `QCA` all satisfy this
    Protocol; a new model only needs a `build_hdh` method, no base class.
    `tests/test_model_protocol.py` checks every registered model against the
    invariants below, so a new model should be added to its registry.

    The HDH returned by `build_hdh` must:

    - create every node with `HDH.add_node`, so each has a type, a time and
      a wire (qubit, bit, or model label);
    - keep each wire a single type (a label used for a classical result is
      never reused as a quantum state);
    - only connect nodes that exist;
    - keep quantum hyperedges free of classical nodes.

    Examples:
        >>> from hdh.models.circuit import Circuit
        >>> isinstance(Circuit(), Model)
        True
    """

    def build_hdh(self) -> HDH:
        """Translate the model's recorded operations into an HDH."""
        ...

build_hdh()

Translate the model's recorded operations into an HDH.

Source code in hdh\models\base.py
31
32
33
def build_hdh(self) -> HDH:
    """Translate the model's recorded operations into an HDH."""
    ...

hdh.models.circuit.Circuit

Gate-model circuit builder: record gates in order, then call build_hdh.

Examples:

>>> c = Circuit()
>>> c.add_instruction("h", [0])
>>> c.add_instruction("cx", [0, 1])
>>> c.add_instruction("measure", [1])
>>> hdh = c.build_hdh()
>>> hdh.get_num_qubits()
2
>>> sorted({hdh.wire_of[n] for n in hdh.nodes})
['c1', 'q0', 'q1']
Source code in hdh\models\circuit.py
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
class Circuit:
    """Gate-model circuit builder: record gates in order, then call `build_hdh`.

    Examples:
        >>> c = Circuit()
        >>> c.add_instruction("h", [0])
        >>> c.add_instruction("cx", [0, 1])
        >>> c.add_instruction("measure", [1])
        >>> hdh = c.build_hdh()
        >>> hdh.get_num_qubits()
        2
        >>> sorted({hdh.wire_of[n] for n in hdh.nodes})
        ['c1', 'q0', 'q1']
    """

    def __init__(self):
        self.instructions: List[
            Tuple[str, List[int], List[int], List[bool], Literal["a", "p"], Optional[List[float]]]
        ] = []  # (name, qubits, bits, modifies_flags, cond_flag, params)

    def add_instruction(
        self,
        name: str,
        qubits: List[int],
        bits: Optional[List[int]] = None,
        modifies_flags: Optional[List[bool]] = None,
        cond_flag: Literal["a", "p"] = "a",
        params: Optional[List[float]] = None,
    ):
        """Append a gate or measurement.

        Args:
            name: Gate name, e.g. ``"h"``, ``"cx"``, ``"rx"`` or ``"measure"``.
            qubits: Qubit indices, in order.
            bits: Classical bits. A measurement defaults to ``bits = qubits``.
            modifies_flags: Per qubit, whether the gate changes its state.
                Defaults to all True.
            cond_flag: ``"p"`` if the gate only happens when a classical
                condition holds (prefer `add_conditional_gate`), else ``"a"``.
            params: Gate parameters such as ``[theta]`` for ``"rx"``; kept on the
                HDH so converters can restore them.

        Examples:
            >>> c = Circuit()
            >>> c.add_instruction("rx", [0], params=[0.5])
            >>> hdh = c.build_hdh()
            >>> list(hdh.gate_params.values())
            [[0.5]]
        """
        name = name.lower()

        if name == "measure":
            modifies_flags = [True] * len(qubits)
        else:
            bits = bits or []
            modifies_flags = modifies_flags or [True] * len(qubits)

        self.instructions.append((name, qubits, bits, modifies_flags, cond_flag, params))

    def add_conditional_gate(
        self,
        classical_bit: int,
        target_qubit: int,
        gate_name: str,
        additional_qubits: Optional[List[int]] = None,
        modifies_flags: Optional[List[bool]] = None,
        params: Optional[List[float]] = None,
    ):
        """Append a gate that only applies when `classical_bit` is 1.

        The bit must already hold a value, typically from an earlier
        ``"measure"``. The gate's output states are marked predicted (``"p"``).

        Args:
            classical_bit: The controlling classical bit.
            target_qubit: Qubit the gate acts on.
            gate_name: Gate name, as in `add_instruction`.
            additional_qubits: Further qubits, for a multi-qubit conditional gate.
            modifies_flags: As in `add_instruction`.
            params: As in `add_instruction`.

        Examples:
            >>> c = Circuit()
            >>> c.add_instruction("h", [0])
            >>> c.add_instruction("measure", [0])
            >>> c.add_conditional_gate(0, 1, "x")
            >>> hdh = c.build_hdh()
            >>> sorted(n for n in hdh.nodes if hdh.node_realisation[n] == "p")
            ['q1_t3']
        """
        gate_name = gate_name.lower()

        # Build the qubit list
        if additional_qubits is None:
            qubits = [target_qubit]
        else:
            qubits = [target_qubit] + additional_qubits

        # Set default modifies_flags
        if modifies_flags is None:
            modifies_flags = [True] * len(qubits)

        # Add the instruction with cond_flag="p" (positive condition)
        # and the classical bit in the bits list
        self.add_instruction(
            name=gate_name,
            qubits=qubits,
            bits=[classical_bit],
            modifies_flags=modifies_flags,
            cond_flag="p",
            params=params,
        )

    def build_hdh(self, hdh_cls=HDH) -> HDH:
        """Translate the recorded instructions into an HDH.

        Qubit ``i`` and bit ``j`` become wires ``"q<i>"`` and ``"c<j>"``. A
        multi-qubit gate spans three timesteps as three hyperedge layers
        (``<name>_stage1``, ``_stage2``, ``_stage3``), so a partitioner can cut
        before, through or after it.

        Args:
            hdh_cls: HDH class to instantiate, for subclasses.
        """
        hdh = hdh_cls()
        qubit_time: Dict[int, int] = {}
        bit_time: Dict[int, int] = {}
        last_gate_input_time: Dict[int, int] = {}

        for name, qargs, cargs, modifies_flags, cond_flag, params in self.instructions:
            # --- Canonicalize inputs ---
            qargs = list(qargs or [])
            if name == "measure":
                cargs = list(cargs) if cargs is not None else qargs.copy()  # 1:1 map
                if len(cargs) != len(qargs):
                    raise ValueError("measure: len(bits) must equal len(qubits)")
                modifies_flags = [True] * len(qargs)
            else:
                cargs = list(cargs or [])
                if modifies_flags is None:
                    modifies_flags = [True] * len(qargs)
                elif len(modifies_flags) != len(qargs):
                    raise ValueError("len(modifies_flags) must equal len(qubits)")

            # Measurements
            if name == "measure":
                for i, qubit in enumerate(qargs):
                    # Use current qubit time (default 0), do NOT advance it here
                    t_in = qubit_time.get(qubit, 0)
                    q_in = f"q{qubit}_t{t_in}"

                    # Check if node already exists - preserve its potential status
                    if q_in not in hdh.nodes:
                        hdh.add_node(f"q{qubit}", t_in, "q", node_real="a")  # Default to actual

                    bit = cargs[i]
                    t_out = t_in + 1              # classical result at next tick
                    c_out = f"c{bit}_t{t_out}"
                    # Classical output is always actual - measurement is unconditional
                    hdh.add_node(f"c{bit}", t_out, "c", node_real=cond_flag)

                    # Measurement hyperedge is always actual - the operation itself is unconditional
                    # (even if measuring a potential quantum state)
                    hdh.add_hyperedge({q_in, c_out}, "c", name="measure", node_real=cond_flag)

                    # Next-free convention for this bit stream
                    bit_time[bit] = t_out + 1

                    # Important: do NOT set qubit_time[qubit] = t_in + k
                    # The quantum wire collapses; keep its last quantum tick unchanged.
                continue

            # Conditional gate handling
            if name != "measure" and cond_flag == "p" and cargs:
                # Supports 1 classical control; extend to many if you like
                ctrl = cargs[0]

                # Ensure times exist
                for q in qargs:
                    if q not in qubit_time:
                        qubit_time[q] = 0  # ← Initialize at t=0
                        last_gate_input_time[q] = 0  # ← Initialize at t=0

                # Classical node must already exist (e.g., produced by a prior measure)
                # bit_time points to "next free" slot; the latest existing node is at t = bit_time-1
                c_latest = bit_time.get(ctrl, 1) - 1
                cnode = f"c{ctrl}_t{c_latest}"
                hdh.add_node(f"c{ctrl}", c_latest, "c", node_real="a")  # Classical node is actual

                edges = []
                for tq in qargs:
                    # gate happens at next tick after both inputs are ready
                    t_in_q = qubit_time[tq]
                    t_gate = max(t_in_q, c_latest) + 1
                    qname = f"q{tq}"

                    # Create input quantum node (actual state before conditional)
                    qin = f"{qname}_t{t_in_q}"
                    hdh.add_node(qname, t_in_q, "q", node_real="a")

                    # Create output quantum node (potential state after conditional)
                    qout = f"{qname}_t{t_gate}"
                    hdh.add_node(qname, t_gate, "q", node_real=cond_flag)

                    # Add quantum hyperedge for wire continuity (potential)
                    q_edge = hdh.add_hyperedge({qin, qout}, "q", name=name, node_real=cond_flag)
                    edges.append(q_edge)

                    # Add classical hyperedge for conditional dependency (potential)
                    c_edge = hdh.add_hyperedge({cnode, qout}, "c", name=name, node_real=cond_flag)
                    edges.append(c_edge)

                    # advance time
                    last_gate_input_time[tq] = t_in_q
                    qubit_time[tq] = t_gate

                # store edge_args for reconstruction/debug
                q_with_time = [(q, qubit_time[q]) for q in qargs]
                c_with_time = [(ctrl, c_latest + 1)]  # next-free convention; adjust if you track exact
                for e in edges:
                    hdh.edge_args[e] = (q_with_time, c_with_time, modifies_flags or [True] * len(qargs))
                    if params:
                        hdh.gate_params[e] = params

                continue

            #Actualized gate (non-conditional)
            for q in qargs:
                if q not in qubit_time:
                    qubit_time[q] = 0  # ← Initialize at t=0
                    last_gate_input_time[q] = 0  # ← Initialize at t=0

            active_times = [qubit_time[q] for q in qargs]
            time_step = max(active_times) + 1 if active_times else 0

            in_nodes: List[str] = []
            out_nodes: List[str] = []

            intermediate_nodes: List[str] = []
            final_nodes: List[str] = []
            post_nodes: List[str] = []

            multi_gate = (name != "measure" and len(qargs) > 1)
            common_start = max((qubit_time.get(q, 0) for q in qargs), default=0) if multi_gate else None

            for i, qubit in enumerate(qargs):
                t_in = qubit_time[qubit]
                qname = f"q{qubit}"
                in_id = f"{qname}_t{t_in}"
                hdh.add_node(qname, t_in, "q", node_real=cond_flag)
                in_nodes.append(in_id)

                # choose timeline
                if multi_gate:
                    t1 = common_start + 1
                    t2 = common_start + 2
                    t3 = common_start + 3

                    # FIX ISSUE #37: Create intermediate nodes INSIDE loop for each qubit
                    mid_id   = f"{qname}_t{t1}"
                    final_id = f"{qname}_t{t2}"
                    post_id  = f"{qname}_t{t3}"

                    hdh.add_node(qname, t1, "q", node_real=cond_flag)
                    hdh.add_node(qname, t2, "q", node_real=cond_flag)
                    hdh.add_node(qname, t3, "q", node_real=cond_flag)

                    intermediate_nodes.append(mid_id)
                    final_nodes.append(final_id)
                    post_nodes.append(post_id)

                    last_gate_input_time[qubit] = t_in
                    qubit_time[qubit] = t3
                else:
                    # Single-qubit gates: don't create nodes here
                    # created by the single-qubit handler below
                    t1 = t_in + 1
                    t2 = t1 + 1
                    t3 = t2 + 1

            edges = []
            if len(qargs) > 1:
                # Multi-qubit gate
                # Stage 1: input → intermediate (1:1)
                for in_node, mid_node in zip(in_nodes, intermediate_nodes):
                    e = hdh.add_hyperedge({in_node, mid_node}, "q", name=f"{name}_stage1", node_real=cond_flag)
                    edges.append(e)

                # Stage 2: full multiqubit edge from intermediate → final
                e2 = hdh.add_hyperedge(set(intermediate_nodes) | set(final_nodes), "q", name=f"{name}_stage2", node_real=cond_flag)
                edges.append(e2)

                # Stage 3: final → post (1:1)
                for f_node, p_node in zip(final_nodes, post_nodes):
                    e = hdh.add_hyperedge({f_node, p_node}, "q", name=f"{name}_stage3", node_real=cond_flag)
                    edges.append(e)

            if name == "measure":
                for i, qubit in enumerate(qargs):
                    t_in = qubit_time.get(qubit, 0)
                    q_in = f"q{qubit}_t{t_in}"
                    hdh.add_node(f"q{qubit}", t_in, "q", node_real=cond_flag)

                    bit = cargs[i]
                    t_out = t_in + 1
                    c_out = f"c{bit}_t{t_out}"
                    hdh.add_node(f"c{bit}", t_out, "c", node_real=cond_flag)

                    hdh.add_hyperedge({q_in, c_out}, "c", name="measure", node_real=cond_flag)
                    bit_time[bit] = t_out + 1
                continue

            if name != "measure":
                for bit in cargs:
                    t = bit_time.get(bit, 0)
                    cname = f"c{bit}"
                    out_id = f"{cname}_t{t + 1}"
                    hdh.add_node(cname, t + 1, "c", node_real=cond_flag)
                    out_nodes.append(out_id)
                    bit_time[bit] = t + 1

            all_nodes = set(in_nodes) | set(out_nodes)
            if all(n.startswith("c") for n in all_nodes):
                edge_type = "c"
            elif any(n.startswith("c") for n in all_nodes):
                edge_type = "c"
            else:
                edge_type = "q"

            if len(qargs) == 1:
                # Single-qubit gate
                for i, qubit in enumerate(qargs):
                    if modifies_flags[i] and name != "measure":
                        # Use current qubit_time
                        t_in = qubit_time[qubit]  
                        t_out = t_in + 1
                        qname = f"q{qubit}"
                        in_id = f"{qname}_t{t_in}"
                        out_id = f"{qname}_t{t_out}"
                        hdh.add_node(qname, t_out, "q", node_real=cond_flag)
                        edge = hdh.add_hyperedge({in_id, out_id}, "q", name=name, node_real=cond_flag)
                        edges.append(edge)
                        # Update time for next gate
                        qubit_time[qubit] = t_out
                        last_gate_input_time[qubit] = t_in

            q_with_time = [(q, qubit_time[q]) for q in qargs]
            c_with_time = [(c, bit_time.get(c, 0)) for c in cargs]
            for edge in edges:
                hdh.edge_args[edge] = (q_with_time, c_with_time, modifies_flags)
                if params:
                    hdh.gate_params[edge] = params

        return hdh

add_conditional_gate(classical_bit, target_qubit, gate_name, additional_qubits=None, modifies_flags=None, params=None)

Append a gate that only applies when classical_bit is 1.

The bit must already hold a value, typically from an earlier "measure". The gate's output states are marked predicted ("p").

Parameters:
  • classical_bit (int) –

    The controlling classical bit.

  • target_qubit (int) –

    Qubit the gate acts on.

  • gate_name (str) –

    Gate name, as in add_instruction.

  • additional_qubits (Optional[List[int]], default: None ) –

    Further qubits, for a multi-qubit conditional gate.

  • modifies_flags (Optional[List[bool]], default: None ) –

    As in add_instruction.

  • params (Optional[List[float]], default: None ) –

    As in add_instruction.

Examples:

>>> c = Circuit()
>>> c.add_instruction("h", [0])
>>> c.add_instruction("measure", [0])
>>> c.add_conditional_gate(0, 1, "x")
>>> hdh = c.build_hdh()
>>> sorted(n for n in hdh.nodes if hdh.node_realisation[n] == "p")
['q1_t3']
Source code in hdh\models\circuit.py
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def add_conditional_gate(
    self,
    classical_bit: int,
    target_qubit: int,
    gate_name: str,
    additional_qubits: Optional[List[int]] = None,
    modifies_flags: Optional[List[bool]] = None,
    params: Optional[List[float]] = None,
):
    """Append a gate that only applies when `classical_bit` is 1.

    The bit must already hold a value, typically from an earlier
    ``"measure"``. The gate's output states are marked predicted (``"p"``).

    Args:
        classical_bit: The controlling classical bit.
        target_qubit: Qubit the gate acts on.
        gate_name: Gate name, as in `add_instruction`.
        additional_qubits: Further qubits, for a multi-qubit conditional gate.
        modifies_flags: As in `add_instruction`.
        params: As in `add_instruction`.

    Examples:
        >>> c = Circuit()
        >>> c.add_instruction("h", [0])
        >>> c.add_instruction("measure", [0])
        >>> c.add_conditional_gate(0, 1, "x")
        >>> hdh = c.build_hdh()
        >>> sorted(n for n in hdh.nodes if hdh.node_realisation[n] == "p")
        ['q1_t3']
    """
    gate_name = gate_name.lower()

    # Build the qubit list
    if additional_qubits is None:
        qubits = [target_qubit]
    else:
        qubits = [target_qubit] + additional_qubits

    # Set default modifies_flags
    if modifies_flags is None:
        modifies_flags = [True] * len(qubits)

    # Add the instruction with cond_flag="p" (positive condition)
    # and the classical bit in the bits list
    self.add_instruction(
        name=gate_name,
        qubits=qubits,
        bits=[classical_bit],
        modifies_flags=modifies_flags,
        cond_flag="p",
        params=params,
    )

add_instruction(name, qubits, bits=None, modifies_flags=None, cond_flag='a', params=None)

Append a gate or measurement.

Parameters:
  • name (str) –

    Gate name, e.g. "h", "cx", "rx" or "measure".

  • qubits (List[int]) –

    Qubit indices, in order.

  • bits (Optional[List[int]], default: None ) –

    Classical bits. A measurement defaults to bits = qubits.

  • modifies_flags (Optional[List[bool]], default: None ) –

    Per qubit, whether the gate changes its state. Defaults to all True.

  • cond_flag (Literal['a', 'p'], default: 'a' ) –

    "p" if the gate only happens when a classical condition holds (prefer add_conditional_gate), else "a".

  • params (Optional[List[float]], default: None ) –

    Gate parameters such as [theta] for "rx"; kept on the HDH so converters can restore them.

Examples:

>>> c = Circuit()
>>> c.add_instruction("rx", [0], params=[0.5])
>>> hdh = c.build_hdh()
>>> list(hdh.gate_params.values())
[[0.5]]
Source code in hdh\models\circuit.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def add_instruction(
    self,
    name: str,
    qubits: List[int],
    bits: Optional[List[int]] = None,
    modifies_flags: Optional[List[bool]] = None,
    cond_flag: Literal["a", "p"] = "a",
    params: Optional[List[float]] = None,
):
    """Append a gate or measurement.

    Args:
        name: Gate name, e.g. ``"h"``, ``"cx"``, ``"rx"`` or ``"measure"``.
        qubits: Qubit indices, in order.
        bits: Classical bits. A measurement defaults to ``bits = qubits``.
        modifies_flags: Per qubit, whether the gate changes its state.
            Defaults to all True.
        cond_flag: ``"p"`` if the gate only happens when a classical
            condition holds (prefer `add_conditional_gate`), else ``"a"``.
        params: Gate parameters such as ``[theta]`` for ``"rx"``; kept on the
            HDH so converters can restore them.

    Examples:
        >>> c = Circuit()
        >>> c.add_instruction("rx", [0], params=[0.5])
        >>> hdh = c.build_hdh()
        >>> list(hdh.gate_params.values())
        [[0.5]]
    """
    name = name.lower()

    if name == "measure":
        modifies_flags = [True] * len(qubits)
    else:
        bits = bits or []
        modifies_flags = modifies_flags or [True] * len(qubits)

    self.instructions.append((name, qubits, bits, modifies_flags, cond_flag, params))

build_hdh(hdh_cls=HDH)

Translate the recorded instructions into an HDH.

Qubit i and bit j become wires "q<i>" and "c<j>". A multi-qubit gate spans three timesteps as three hyperedge layers (<name>_stage1, _stage2, _stage3), so a partitioner can cut before, through or after it.

Parameters:
  • hdh_cls –

    HDH class to instantiate, for subclasses.

Source code in hdh\models\circuit.py
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
def build_hdh(self, hdh_cls=HDH) -> HDH:
    """Translate the recorded instructions into an HDH.

    Qubit ``i`` and bit ``j`` become wires ``"q<i>"`` and ``"c<j>"``. A
    multi-qubit gate spans three timesteps as three hyperedge layers
    (``<name>_stage1``, ``_stage2``, ``_stage3``), so a partitioner can cut
    before, through or after it.

    Args:
        hdh_cls: HDH class to instantiate, for subclasses.
    """
    hdh = hdh_cls()
    qubit_time: Dict[int, int] = {}
    bit_time: Dict[int, int] = {}
    last_gate_input_time: Dict[int, int] = {}

    for name, qargs, cargs, modifies_flags, cond_flag, params in self.instructions:
        # --- Canonicalize inputs ---
        qargs = list(qargs or [])
        if name == "measure":
            cargs = list(cargs) if cargs is not None else qargs.copy()  # 1:1 map
            if len(cargs) != len(qargs):
                raise ValueError("measure: len(bits) must equal len(qubits)")
            modifies_flags = [True] * len(qargs)
        else:
            cargs = list(cargs or [])
            if modifies_flags is None:
                modifies_flags = [True] * len(qargs)
            elif len(modifies_flags) != len(qargs):
                raise ValueError("len(modifies_flags) must equal len(qubits)")

        # Measurements
        if name == "measure":
            for i, qubit in enumerate(qargs):
                # Use current qubit time (default 0), do NOT advance it here
                t_in = qubit_time.get(qubit, 0)
                q_in = f"q{qubit}_t{t_in}"

                # Check if node already exists - preserve its potential status
                if q_in not in hdh.nodes:
                    hdh.add_node(f"q{qubit}", t_in, "q", node_real="a")  # Default to actual

                bit = cargs[i]
                t_out = t_in + 1              # classical result at next tick
                c_out = f"c{bit}_t{t_out}"
                # Classical output is always actual - measurement is unconditional
                hdh.add_node(f"c{bit}", t_out, "c", node_real=cond_flag)

                # Measurement hyperedge is always actual - the operation itself is unconditional
                # (even if measuring a potential quantum state)
                hdh.add_hyperedge({q_in, c_out}, "c", name="measure", node_real=cond_flag)

                # Next-free convention for this bit stream
                bit_time[bit] = t_out + 1

                # Important: do NOT set qubit_time[qubit] = t_in + k
                # The quantum wire collapses; keep its last quantum tick unchanged.
            continue

        # Conditional gate handling
        if name != "measure" and cond_flag == "p" and cargs:
            # Supports 1 classical control; extend to many if you like
            ctrl = cargs[0]

            # Ensure times exist
            for q in qargs:
                if q not in qubit_time:
                    qubit_time[q] = 0  # ← Initialize at t=0
                    last_gate_input_time[q] = 0  # ← Initialize at t=0

            # Classical node must already exist (e.g., produced by a prior measure)
            # bit_time points to "next free" slot; the latest existing node is at t = bit_time-1
            c_latest = bit_time.get(ctrl, 1) - 1
            cnode = f"c{ctrl}_t{c_latest}"
            hdh.add_node(f"c{ctrl}", c_latest, "c", node_real="a")  # Classical node is actual

            edges = []
            for tq in qargs:
                # gate happens at next tick after both inputs are ready
                t_in_q = qubit_time[tq]
                t_gate = max(t_in_q, c_latest) + 1
                qname = f"q{tq}"

                # Create input quantum node (actual state before conditional)
                qin = f"{qname}_t{t_in_q}"
                hdh.add_node(qname, t_in_q, "q", node_real="a")

                # Create output quantum node (potential state after conditional)
                qout = f"{qname}_t{t_gate}"
                hdh.add_node(qname, t_gate, "q", node_real=cond_flag)

                # Add quantum hyperedge for wire continuity (potential)
                q_edge = hdh.add_hyperedge({qin, qout}, "q", name=name, node_real=cond_flag)
                edges.append(q_edge)

                # Add classical hyperedge for conditional dependency (potential)
                c_edge = hdh.add_hyperedge({cnode, qout}, "c", name=name, node_real=cond_flag)
                edges.append(c_edge)

                # advance time
                last_gate_input_time[tq] = t_in_q
                qubit_time[tq] = t_gate

            # store edge_args for reconstruction/debug
            q_with_time = [(q, qubit_time[q]) for q in qargs]
            c_with_time = [(ctrl, c_latest + 1)]  # next-free convention; adjust if you track exact
            for e in edges:
                hdh.edge_args[e] = (q_with_time, c_with_time, modifies_flags or [True] * len(qargs))
                if params:
                    hdh.gate_params[e] = params

            continue

        #Actualized gate (non-conditional)
        for q in qargs:
            if q not in qubit_time:
                qubit_time[q] = 0  # ← Initialize at t=0
                last_gate_input_time[q] = 0  # ← Initialize at t=0

        active_times = [qubit_time[q] for q in qargs]
        time_step = max(active_times) + 1 if active_times else 0

        in_nodes: List[str] = []
        out_nodes: List[str] = []

        intermediate_nodes: List[str] = []
        final_nodes: List[str] = []
        post_nodes: List[str] = []

        multi_gate = (name != "measure" and len(qargs) > 1)
        common_start = max((qubit_time.get(q, 0) for q in qargs), default=0) if multi_gate else None

        for i, qubit in enumerate(qargs):
            t_in = qubit_time[qubit]
            qname = f"q{qubit}"
            in_id = f"{qname}_t{t_in}"
            hdh.add_node(qname, t_in, "q", node_real=cond_flag)
            in_nodes.append(in_id)

            # choose timeline
            if multi_gate:
                t1 = common_start + 1
                t2 = common_start + 2
                t3 = common_start + 3

                # FIX ISSUE #37: Create intermediate nodes INSIDE loop for each qubit
                mid_id   = f"{qname}_t{t1}"
                final_id = f"{qname}_t{t2}"
                post_id  = f"{qname}_t{t3}"

                hdh.add_node(qname, t1, "q", node_real=cond_flag)
                hdh.add_node(qname, t2, "q", node_real=cond_flag)
                hdh.add_node(qname, t3, "q", node_real=cond_flag)

                intermediate_nodes.append(mid_id)
                final_nodes.append(final_id)
                post_nodes.append(post_id)

                last_gate_input_time[qubit] = t_in
                qubit_time[qubit] = t3
            else:
                # Single-qubit gates: don't create nodes here
                # created by the single-qubit handler below
                t1 = t_in + 1
                t2 = t1 + 1
                t3 = t2 + 1

        edges = []
        if len(qargs) > 1:
            # Multi-qubit gate
            # Stage 1: input → intermediate (1:1)
            for in_node, mid_node in zip(in_nodes, intermediate_nodes):
                e = hdh.add_hyperedge({in_node, mid_node}, "q", name=f"{name}_stage1", node_real=cond_flag)
                edges.append(e)

            # Stage 2: full multiqubit edge from intermediate → final
            e2 = hdh.add_hyperedge(set(intermediate_nodes) | set(final_nodes), "q", name=f"{name}_stage2", node_real=cond_flag)
            edges.append(e2)

            # Stage 3: final → post (1:1)
            for f_node, p_node in zip(final_nodes, post_nodes):
                e = hdh.add_hyperedge({f_node, p_node}, "q", name=f"{name}_stage3", node_real=cond_flag)
                edges.append(e)

        if name == "measure":
            for i, qubit in enumerate(qargs):
                t_in = qubit_time.get(qubit, 0)
                q_in = f"q{qubit}_t{t_in}"
                hdh.add_node(f"q{qubit}", t_in, "q", node_real=cond_flag)

                bit = cargs[i]
                t_out = t_in + 1
                c_out = f"c{bit}_t{t_out}"
                hdh.add_node(f"c{bit}", t_out, "c", node_real=cond_flag)

                hdh.add_hyperedge({q_in, c_out}, "c", name="measure", node_real=cond_flag)
                bit_time[bit] = t_out + 1
            continue

        if name != "measure":
            for bit in cargs:
                t = bit_time.get(bit, 0)
                cname = f"c{bit}"
                out_id = f"{cname}_t{t + 1}"
                hdh.add_node(cname, t + 1, "c", node_real=cond_flag)
                out_nodes.append(out_id)
                bit_time[bit] = t + 1

        all_nodes = set(in_nodes) | set(out_nodes)
        if all(n.startswith("c") for n in all_nodes):
            edge_type = "c"
        elif any(n.startswith("c") for n in all_nodes):
            edge_type = "c"
        else:
            edge_type = "q"

        if len(qargs) == 1:
            # Single-qubit gate
            for i, qubit in enumerate(qargs):
                if modifies_flags[i] and name != "measure":
                    # Use current qubit_time
                    t_in = qubit_time[qubit]  
                    t_out = t_in + 1
                    qname = f"q{qubit}"
                    in_id = f"{qname}_t{t_in}"
                    out_id = f"{qname}_t{t_out}"
                    hdh.add_node(qname, t_out, "q", node_real=cond_flag)
                    edge = hdh.add_hyperedge({in_id, out_id}, "q", name=name, node_real=cond_flag)
                    edges.append(edge)
                    # Update time for next gate
                    qubit_time[qubit] = t_out
                    last_gate_input_time[qubit] = t_in

        q_with_time = [(q, qubit_time[q]) for q in qargs]
        c_with_time = [(c, bit_time.get(c, 0)) for c in cargs]
        for edge in edges:
            hdh.edge_args[edge] = (q_with_time, c_with_time, modifies_flags)
            if params:
                hdh.gate_params[edge] = params

    return hdh

hdh.models.mbqc.MBQC

Measurement-based quantum computing (MBQC) pattern builder.

Record N (prepare), E (entangle), M (measure) and C (classical correction) operations with add_operation, then call build_hdh. Labels are free-form and become the HDH's wires; keep each label to one type (never reuse a measurement result's label as a quantum state).

Examples:

>>> m = MBQC()
>>> m.add_operation("N", [], "a")
>>> m.add_operation("N", [], "b")
>>> m.add_operation("E", ["a", "b"], "b")
>>> m.add_operation("M", ["a"], "m_a")
>>> hdh = m.build_hdh()
>>> sorted({hdh.wire_of[n] for n in hdh.nodes if hdh.node_types[n] == "q"})
['a', 'b']
Source code in hdh\models\mbqc.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class MBQC:
    """Measurement-based quantum computing (MBQC) pattern builder.

    Record N (prepare), E (entangle), M (measure) and C (classical
    correction) operations with `add_operation`, then call `build_hdh`.
    Labels are free-form and become the HDH's wires; keep each label to one
    type (never reuse a measurement result's label as a quantum state).

    Examples:
        >>> m = MBQC()
        >>> m.add_operation("N", [], "a")
        >>> m.add_operation("N", [], "b")
        >>> m.add_operation("E", ["a", "b"], "b")
        >>> m.add_operation("M", ["a"], "m_a")
        >>> hdh = m.build_hdh()
        >>> sorted({hdh.wire_of[n] for n in hdh.nodes if hdh.node_types[n] == "q"})
        ['a', 'b']
    """

    def __init__(self, hdh_cls=HDH):
        self.pattern = []  # (op_type, A, b)
        self.hdh_cls = hdh_cls

    def add_operation(self, op_type: str, A: List[str], b: str):
        """Append one N, E, M or C operation.

        Args:
            op_type: ``"N"``, ``"E"``, ``"M"`` or ``"C"`` (case-insensitive).
            A: Labels the operation reads; empty for ``"N"``.
            b: Label it produces. For ``"E"``, reuse one of the entangled labels.
        """
        self.pattern.append((op_type.upper(), A, b))

    def build_hdh(self) -> HDH:
        """Translate the recorded pattern into an HDH, one timestep per operation."""
        hdh = self.hdh_cls()
        time_map = {}
        current_time = 0

        for op_type, A, b in self.pattern:
            in_nodes = set()
            out_nodes = set()
            all_nodes = A + [b]

            # Assign time steps
            op_time = current_time
            current_time += 1

            for x in A:
                t = time_map.get(x, 0)
                in_nodes.add(hdh.add_node(x, t, self._node_type(op_type, input=True)))

            out_nodes.add(hdh.add_node(b, op_time, self._node_type(op_type, input=False)))
            time_map[b] = op_time

            edge_nodes = in_nodes | out_nodes
            hdh.add_hyperedge(edge_nodes, self._edge_type(op_type), name=op_type.lower())

        return hdh

    def _node_type(self, op_type, input=False):
        """Map an NEMC op type + input/output position to a `sigma` value."""
        if op_type == "N":
            return "c" if input else "q"
        if op_type == "E":
            return "q"
        if op_type == "M":
            return "q" if input else "c"
        if op_type == "C":
            return "c"

    def _edge_type(self, op_type):
        """Map an NEMC op type to a `tau` value: only E is quantum."""
        return "q" if op_type == "E" else "c"

add_operation(op_type, A, b)

Append one N, E, M or C operation.

Parameters:
  • op_type (str) –

    "N", "E", "M" or "C" (case-insensitive).

  • A (List[str]) –

    Labels the operation reads; empty for "N".

  • b (str) –

    Label it produces. For "E", reuse one of the entangled labels.

Source code in hdh\models\mbqc.py
33
34
35
36
37
38
39
40
41
def add_operation(self, op_type: str, A: List[str], b: str):
    """Append one N, E, M or C operation.

    Args:
        op_type: ``"N"``, ``"E"``, ``"M"`` or ``"C"`` (case-insensitive).
        A: Labels the operation reads; empty for ``"N"``.
        b: Label it produces. For ``"E"``, reuse one of the entangled labels.
    """
    self.pattern.append((op_type.upper(), A, b))

build_hdh()

Translate the recorded pattern into an HDH, one timestep per operation.

Source code in hdh\models\mbqc.py
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def build_hdh(self) -> HDH:
    """Translate the recorded pattern into an HDH, one timestep per operation."""
    hdh = self.hdh_cls()
    time_map = {}
    current_time = 0

    for op_type, A, b in self.pattern:
        in_nodes = set()
        out_nodes = set()
        all_nodes = A + [b]

        # Assign time steps
        op_time = current_time
        current_time += 1

        for x in A:
            t = time_map.get(x, 0)
            in_nodes.add(hdh.add_node(x, t, self._node_type(op_type, input=True)))

        out_nodes.add(hdh.add_node(b, op_time, self._node_type(op_type, input=False)))
        time_map[b] = op_time

        edge_nodes = in_nodes | out_nodes
        hdh.add_hyperedge(edge_nodes, self._edge_type(op_type), name=op_type.lower())

    return hdh

hdh.models.qca.QCA

Quantum cellular automaton builder: cells updated from their neighbors for a number of steps, then optionally measured.

Parameters:
  • topology –

    Each cell label (any string) -> the neighbor labels its update reads.

  • measurements –

    Cells to measure after the last step.

  • steps –

    Number of update steps.

  • hdh_cls –

    HDH class to instantiate, for subclasses.

Examples:

>>> qca = QCA(topology={"A": ["B"], "B": ["A"]}, measurements={"A"}, steps=1)
>>> sorted(qca.build_hdh().nodes)
['A_t0', 'A_t1', 'B_t0', 'B_t1', 'c_A_t2']
Source code in hdh\models\qca.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class QCA:
    """Quantum cellular automaton builder: cells updated from their neighbors for
    a number of steps, then optionally measured.

    Args:
        topology: Each cell label (any string) -> the neighbor labels its
            update reads.
        measurements: Cells to measure after the last step.
        steps: Number of update steps.
        hdh_cls: HDH class to instantiate, for subclasses.

    Examples:
        >>> qca = QCA(topology={"A": ["B"], "B": ["A"]}, measurements={"A"}, steps=1)
        >>> sorted(qca.build_hdh().nodes)
        ['A_t0', 'A_t1', 'B_t0', 'B_t1', 'c_A_t2']
    """

    def __init__(self, topology, measurements, steps, hdh_cls=HDH):
        self.topology = topology
        self.measurements = measurements
        self.steps = steps
        self.hdh_cls = hdh_cls

    def build_hdh(self) -> HDH:
        """Build the HDH: per step, one hyperedge per cell joining its and its
        neighbors' previous states to its new state; then one measurement
        hyperedge per measured cell.
        """
        hdh = self.hdh_cls()
        time_map = {node: 0 for node in self.topology}

        for t in range(1, self.steps + 1):
            for node, neighbors in self.topology.items():
                inputs = []
                for n in neighbors + [node]:
                    inputs.append(hdh.add_node(n, time_map[n], "q"))

                out_node = hdh.add_node(node, t, "q")
                hdh.add_hyperedge(frozenset(inputs + [out_node]), "q", name="update")
                time_map[node] = t

        # Add measurement edges
        for node in self.measurements:
            t_meas = self.steps + 1  # important!
            out_node = hdh.node_id(node, self.steps)
            c_node = hdh.add_node(self._classical_label(node), t_meas, "c")
            hdh.add_hyperedge(frozenset({out_node, c_node}), "c", name="measure")

        return hdh

    @staticmethod
    def _classical_label(cell: str) -> str:
        """Label for the classical bit a measured cell writes to.

        Cells named ``q<int>`` keep the circuit convention (``q3`` -> ``c3``)
        so converters recognise the bit; any other name gets a ``c_`` prefix
        (``"A"`` -> ``"c_A"``).
        """
        m = re.fullmatch(r"q(\d+)", cell)
        return f"c{m.group(1)}" if m else f"c_{cell}"

build_hdh()

Build the HDH: per step, one hyperedge per cell joining its and its neighbors' previous states to its new state; then one measurement hyperedge per measured cell.

Source code in hdh\models\qca.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def build_hdh(self) -> HDH:
    """Build the HDH: per step, one hyperedge per cell joining its and its
    neighbors' previous states to its new state; then one measurement
    hyperedge per measured cell.
    """
    hdh = self.hdh_cls()
    time_map = {node: 0 for node in self.topology}

    for t in range(1, self.steps + 1):
        for node, neighbors in self.topology.items():
            inputs = []
            for n in neighbors + [node]:
                inputs.append(hdh.add_node(n, time_map[n], "q"))

            out_node = hdh.add_node(node, t, "q")
            hdh.add_hyperedge(frozenset(inputs + [out_node]), "q", name="update")
            time_map[node] = t

    # Add measurement edges
    for node in self.measurements:
        t_meas = self.steps + 1  # important!
        out_node = hdh.node_id(node, self.steps)
        c_node = hdh.add_node(self._classical_label(node), t_meas, "c")
        hdh.add_hyperedge(frozenset({out_node, c_node}), "c", name="measure")

    return hdh

hdh.models.qw.QW

Discrete-time quantum walk builder: coin and shift steps, then measurement.

add_coin and add_shift return the label of the new walker state, to pass to the next step.

Examples:

>>> w = QW()
>>> state = w.add_shift(w.add_coin("q0"))
>>> w.add_measurement(state, "m0")
>>> sorted(w.build_hdh().nodes)
['m0_t3', 'q0_t0', 'q1_t1', 'q2_t2']
Source code in hdh\models\qw.py
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class QW:
    """Discrete-time quantum walk builder: coin and shift steps, then measurement.

    `add_coin` and `add_shift` return the label of the new walker state, to
    pass to the next step.

    Examples:
        >>> w = QW()
        >>> state = w.add_shift(w.add_coin("q0"))
        >>> w.add_measurement(state, "m0")
        >>> sorted(w.build_hdh().nodes)
        ['m0_t3', 'q0_t0', 'q1_t1', 'q2_t2']
    """

    def __init__(self, hdh_cls=HDH):
        self.steps = []  # (type, a, b)
        self.hdh_cls = hdh_cls
        self.qubit_counter = 0  # For auto-generating digit-only qubit IDs

    def _new_qubit_id(self):
        """Generate the next auto-incrementing walker-state label."""
        self.qubit_counter += 1
        return f"q{self.qubit_counter}"

    def add_coin(self, a: str):
        """Apply a coin operation to walker state `a`; returns the new state label."""
        a_prime = self._new_qubit_id()
        self.steps.append(("K", a, a_prime))
        return a_prime

    def add_shift(self, a_prime: str):
        """Apply a shift operation to walker state `a_prime`; returns the new state label."""
        b = self._new_qubit_id()
        self.steps.append(("R", a_prime, b))
        return b

    def add_measurement(self, a: str, b: str):
        """Measure walker state `a` into classical label `b`, which must not already
        be used for a quantum state.
        """
        self.steps.append(("M", a, b))

    def build_hdh(self) -> HDH:
        """Translate the recorded coin/shift/measurement steps into an HDH."""
        hdh = self.hdh_cls()
        time_map: Dict[str, int] = {}

        for step_index, (op_type, a, b) in enumerate(self.steps):
            in_time = time_map.get(a, 0)
            out_time = in_time + 1

            in_type = "q"
            out_type = "q" if op_type in {"K", "R"} else "c"
            edge_type = "q" if op_type in {"K", "R"} else "c"

            in_id = hdh.add_node(a, in_time, in_type)
            out_id = hdh.add_node(b, out_time, out_type)
            hdh.add_hyperedge({in_id, out_id}, edge_type, name=op_type.lower())

            time_map[b] = out_time  # set output time
        return hdh

add_coin(a)

Apply a coin operation to walker state a; returns the new state label.

Source code in hdh\models\qw.py
31
32
33
34
35
def add_coin(self, a: str):
    """Apply a coin operation to walker state `a`; returns the new state label."""
    a_prime = self._new_qubit_id()
    self.steps.append(("K", a, a_prime))
    return a_prime

add_measurement(a, b)

Measure walker state a into classical label b, which must not already be used for a quantum state.

Source code in hdh\models\qw.py
43
44
45
46
47
def add_measurement(self, a: str, b: str):
    """Measure walker state `a` into classical label `b`, which must not already
    be used for a quantum state.
    """
    self.steps.append(("M", a, b))

add_shift(a_prime)

Apply a shift operation to walker state a_prime; returns the new state label.

Source code in hdh\models\qw.py
37
38
39
40
41
def add_shift(self, a_prime: str):
    """Apply a shift operation to walker state `a_prime`; returns the new state label."""
    b = self._new_qubit_id()
    self.steps.append(("R", a_prime, b))
    return b

build_hdh()

Translate the recorded coin/shift/measurement steps into an HDH.

Source code in hdh\models\qw.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def build_hdh(self) -> HDH:
    """Translate the recorded coin/shift/measurement steps into an HDH."""
    hdh = self.hdh_cls()
    time_map: Dict[str, int] = {}

    for step_index, (op_type, a, b) in enumerate(self.steps):
        in_time = time_map.get(a, 0)
        out_time = in_time + 1

        in_type = "q"
        out_type = "q" if op_type in {"K", "R"} else "c"
        edge_type = "q" if op_type in {"K", "R"} else "c"

        in_id = hdh.add_node(a, in_time, in_type)
        out_id = hdh.add_node(b, out_time, out_type)
        hdh.add_hyperedge({in_id, out_id}, edge_type, name=op_type.lower())

        time_map[b] = out_time  # set output time
    return hdh

Visualization

hdh.visualize.plot_hdh(hdh, save_path='hdh_plot.svg')

Plot an HDH with time on the x-axis and one row per wire.

Circuit wires q<i> and c<i> share row i; any other wire gets a row of its own.

Parameters:
  • hdh –

    The HDH to plot.

  • save_path –

    File to save to (the extension sets the format), or None to show the plot instead.

Examples:

>>> from hdh.models.circuit import Circuit
>>> c = Circuit()
>>> c.add_instruction("cx", [0, 1])
>>> plot_hdh(c.build_hdh(), save_path="cx.svg")
Source code in hdh\visualize.py
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def plot_hdh(hdh, save_path="hdh_plot.svg"):
    """Plot an HDH with time on the x-axis and one row per wire.

    Circuit wires ``q<i>`` and ``c<i>`` share row ``i``; any other wire gets a
    row of its own.

    Args:
        hdh: The HDH to plot.
        save_path: File to save to (the extension sets the format), or
            ``None`` to show the plot instead.

    Examples:
        >>> from hdh.models.circuit import Circuit
        >>> c = Circuit()
        >>> c.add_instruction("cx", [0, 1])
        >>> plot_hdh(c.build_hdh(), save_path="cx.svg")  # doctest: +SKIP
    """
    nodes = list(hdh.nodes)
    edges = [tuple(e) for e in hdh.hyperedges]

    if not nodes:
        print("Nothing to plot: the HDH has no nodes.")
        return

    # One row per wire, read from the HDH rather than parsed from node IDs.
    # Circuit wires q<i> and c<i> share row i (a qubit and the bit it is
    # measured into); any other wire, e.g. an MBQC label, gets its own row.
    def _row_key(wire):
        m = re.fullmatch(r"[qc](\d+)", wire)
        return int(m.group(1)) if m else wire

    keys = {_row_key(hdh.wire_of[n]) for n in nodes}
    numbered = sorted(k for k in keys if isinstance(k, int))
    named = sorted(k for k in keys if not isinstance(k, int))
    row_order = numbered + named
    row_of = {key: len(row_order) - 1 - i for i, key in enumerate(row_order)}  # first row on top

    node_positions = {
        n: (hdh.time_map[n], row_of[_row_key(hdh.wire_of[n])]) for n in nodes
    }
    timestep_ticks = sorted({hdh.time_map[n] for n in nodes})

    fig, ax = plt.subplots(figsize=(10, 4))
    ax.set_xlabel("Timestep",fontsize=16)
    ax.set_ylabel("Qubit/Clbit Index",fontsize=16)
    ax.set_xticks(timestep_ticks)
    ax.set_yticks([row_of[k] for k in row_order])
    ax.set_yticklabels([str(k) for k in row_order])
    ax.set_ylim(-1, len(row_order))

    involved_nodes = set()
    for edge in edges:
        involved_nodes.update(edge)

    for node in involved_nodes:
        if node in node_positions:
            x, y = node_positions[node]
            node_type = hdh.node_types.get(node, "q")
            is_predicted = hdh.node_realisation.get(node, "a") == "p"
            color = {
                "q": "black",
                "ctrl": "black",
                "c": "orange"
            }.get(node_type, "black")

            face_color = "white" if is_predicted else color
            ax.plot(x, y, 'o', markersize=10,
                    markerfacecolor=face_color,
                    markeredgecolor=color,
                    markeredgewidth=2,
                    linestyle='--' if is_predicted else '-')
            ax.text(x, y + 0.15, node, ha='center')

    seen_pairs = set()
    for edge in edges:
        edge_nodes = [n for n in edge if n in node_positions]

        edge_type = hdh.hyperedge_types.get(frozenset(edge))
        if edge_type is None:
            node_types = [hdh.node_types.get(n, "q") for n in edge]
            edge_type = "c" if all(t == "c" for t in node_types) else "q"

        color = "orange" if edge_type == "c" else "black"
        is_predicted = hdh.hyperedge_realisation.get(frozenset(edge), "a") == "p"
        line_style = '--' if is_predicted else '-'

        for i in range(len(edge_nodes)):
            for j in range(i + 1, len(edge_nodes)):
                n1, n2 = edge_nodes[i], edge_nodes[j]
                t1, t2 = hdh.time_map[n1], hdh.time_map[n2]

                if t1 == t2:
                    continue

                type1 = hdh.node_types.get(n1, "q")
                type2 = hdh.node_types.get(n2, "q")
                if type1 == "ctrl" and type2 == "ctrl":
                    continue

                if t1 > t2:
                    n1, n2 = n2, n1
                    t1, t2 = t2, t1

                pair = (n1, n2)
                if pair in seen_pairs:
                    continue
                seen_pairs.add(pair)

                x0, y0 = node_positions[n1]
                x1, y1 = node_positions[n2]

                if edge_type == "c":
                    dx = x1 - x0
                    dy = y1 - y0
                    dist = np.hypot(dx, dy)
                    if dist == 0:
                        continue
                    # Calculate perpendicular unit vector
                    nx_vec = -dy / dist
                    ny_vec = dx / dist
                    # Offset distance for double line
                    offset = 0.05
                    # Draw two parallel lines
                    ax.plot([x0 + offset * nx_vec, x1 + offset * nx_vec], 
                            [y0 + offset * ny_vec, y1 + offset * ny_vec], 
                            color=color, linewidth=1, linestyle=line_style)
                    ax.plot([x0 - offset * nx_vec, x1 - offset * nx_vec], 
                            [y0 - offset * ny_vec, y1 - offset * ny_vec], 
                            color=color, linewidth=1, linestyle=line_style)
                else:
                    ax.plot([x0, x1], [y0, y1], color=color, linewidth=1.5, linestyle=line_style)

    if save_path:
        ext = os.path.splitext(save_path)[1].lower()
        if ext in [".png", ".jpg"]:
            plt.savefig(save_path, dpi=600, bbox_inches='tight')
        else:
            plt.savefig(save_path, bbox_inches='tight')
        print(f"Plot saved to: {save_path}")
    else:
        plt.show()

Converters

hdh.converters.qiskit_converter.from_qiskit(qc)

Convert a Qiskit QuantumCircuit to an HDH.

Supports standard gates, measurement, and if_test blocks conditioned on a single classical bit equal to 1.

Raises:
  • NotImplementedError –

    For unsupported operations or conditions.

Examples:

>>> from qiskit import QuantumCircuit
>>> qc = QuantumCircuit(2, 1)
>>> _ = qc.h(0); _ = qc.cx(0, 1); _ = qc.measure(1, 0)
>>> from_qiskit(qc).get_num_qubits()
2
Source code in hdh\converters\qiskit_converter.py
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def from_qiskit(qc: QuantumCircuit) -> HDH:
    """Convert a Qiskit `QuantumCircuit` to an HDH.

    Supports standard gates, measurement, and `if_test` blocks conditioned on
    a single classical bit equal to 1.

    Raises:
        NotImplementedError: For unsupported operations or conditions.

    Examples:
        >>> from qiskit import QuantumCircuit
        >>> qc = QuantumCircuit(2, 1)
        >>> _ = qc.h(0); _ = qc.cx(0, 1); _ = qc.measure(1, 0)
        >>> from_qiskit(qc).get_num_qubits()
        2
    """
    circuit = Circuit()

    for item in qc.data:
        instr, qargs, cargs = item.operation, item.qubits, item.clbits
        # Skip metadata instructions
        if instr.name in {"barrier", "snapshot", "delay", "label"}:
            continue

        # Get indices
        q_indices = [qc.qubits.index(q) for q in qargs]
        c_indices = [qc.clbits.index(c) for c in cargs]

        # Handle IfElseOp
        if isinstance(instr, IfElseOp):
            _process_if_else_op(qc, instr, circuit)
            continue

        # Handle standard instructions
        if instr.name == "measure":
            circuit.add_instruction("measure", q_indices, None)
        else:
            modifies_flags = [True] * len(q_indices)
            raw_params = getattr(instr, "params", None) or []
            try:
                params = [float(p) for p in raw_params] or None
            except (TypeError, ValueError):
                # Unbound symbolic Parameter objects can't be stored as floats.
                params = None
            circuit.add_instruction(instr.name, q_indices, c_indices, modifies_flags, params=params)

    return circuit.build_hdh()

hdh.converters.qiskit_converter.to_qiskit(hdh)

Convert an HDH back into a Qiskit QuantumCircuit.

Each multi-qubit gate is rebuilt from its _stage2 hyperedge. Gate parameters come from HDH.gate_params when recorded (e.g. by from_qiskit) and default to 0 otherwise.

Examples:

>>> from qiskit import QuantumCircuit
>>> qc = QuantumCircuit(2)
>>> _ = qc.h(0); _ = qc.cx(0, 1)
>>> [inst.operation.name for inst in to_qiskit(from_qiskit(qc)).data]
['h', 'cx']
Source code in hdh\converters\qiskit_converter.py
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
def to_qiskit(hdh: HDH) -> QuantumCircuit:
    """Convert an HDH back into a Qiskit `QuantumCircuit`.

    Each multi-qubit gate is rebuilt from its ``_stage2`` hyperedge. Gate
    parameters come from `HDH.gate_params` when recorded (e.g. by
    `from_qiskit`) and default to 0 otherwise.

    Examples:
        >>> from qiskit import QuantumCircuit
        >>> qc = QuantumCircuit(2)
        >>> _ = qc.h(0); _ = qc.cx(0, 1)
        >>> [inst.operation.name for inst in to_qiskit(from_qiskit(qc)).data]
        ['h', 'cx']
    """
    qubit_indices: Set[int] = set()
    bit_indices:   Set[int] = set()

    for node_id in hdh.nodes:
        q = _parse_qubit(node_id)
        if q is not None:
            qubit_indices.add(q)
            continue
        c = _parse_cbit(node_id)
        if c is not None:
            bit_indices.add(c)

    # Map global qubit/bit indices to a dense local range so a partition holding
    # e.g. qubits {2, 3} produces a 2-qubit circuit, not a 4-qubit one.
    qubit_map = {q: i for i, q in enumerate(sorted(qubit_indices))}
    bit_map   = {c: i for i, c in enumerate(sorted(bit_indices))}
    n_qubits  = len(qubit_map)
    n_bits    = len(bit_map)

    if n_qubits == 0:
        return QuantumCircuit()

    qc = QuantumCircuit(n_qubits, n_bits) if n_bits > 0 else QuantumCircuit(n_qubits)

    records: List[tuple] = []

    for edge in hdh.hyperedges:
        raw_name  = hdh.gate_name.get(edge, "")
        edge_type = hdh.hyperedge_types.get(edge, "q")

        if raw_name == "measure":
            q_nodes = sorted(
                (n for n in edge if hdh.node_types.get(n) == "q"),
                key=lambda n: hdh.time_map.get(n, 0),
            )
            c_nodes = sorted(
                (n for n in edge if hdh.node_types.get(n) == "c"),
                key=lambda n: hdh.time_map.get(n, 0),
            )
            if not q_nodes or not c_nodes:
                continue
            q_idx = _parse_qubit(q_nodes[0])
            c_idx = _parse_cbit(c_nodes[0])
            if q_idx is None or c_idx is None or q_idx not in qubit_map or c_idx not in bit_map:
                continue
            sort_time = hdh.time_map.get(c_nodes[0], 0)
            records.append((sort_time, "measure", [qubit_map[q_idx]], [bit_map[c_idx]], False, None))
            continue

        # Stages 1 and 3 only carry wire continuity; stage 2 holds the real gate.
        if raw_name.endswith("_stage1") or raw_name.endswith("_stage3"):
            continue

        # Classical edges of a conditional gate duplicate its quantum edge.
        if edge_type == "c":
            continue

        args = hdh.edge_args.get(edge)
        if args is None:
            continue

        q_with_time, c_with_time, _ = args
        actual_name = raw_name[:-7] if raw_name.endswith("_stage2") else raw_name
        q_indices   = [qubit_map[q] for q, _ in q_with_time if q in qubit_map]
        c_indices   = [bit_map[c] for c, _ in c_with_time if c in bit_map]
        sort_time   = min((t for _, t in q_with_time), default=0)
        is_cond     = hdh.hyperedge_realisation.get(edge) == "p"

        records.append((sort_time, actual_name, q_indices, c_indices, is_cond, hdh.gate_params.get(edge)))

    records.sort(key=lambda r: r[0])

    # A multi-qubit gate contributes one record per qubit wire; keep one.
    seen: Set[tuple] = set()
    unique_records: List[tuple] = []
    for rec in records:
        sort_time, name, q_indices, c_indices, is_cond, params = rec
        dedup_key = (sort_time, name, tuple(q_indices))
        if dedup_key not in seen:
            seen.add(dedup_key)
            unique_records.append(rec)

    for sort_time, name, q_indices, c_indices, is_cond, params in unique_records:
        if name == "measure":
            for qi, ci in zip(q_indices, c_indices):
                qc.measure(qi, ci)
            continue

        gate = _make_gate(name, params)
        if gate is None:
            print(f"[WARNING] Unrecognised gate '{name}' at t={sort_time} on qubits {q_indices} - skipped.")
            continue

        if any(qi >= n_qubits or qi < 0 for qi in q_indices):
            print(f"[WARNING] Gate '{name}' references out-of-range qubits {q_indices} - skipped.")
            continue

        if gate.num_qubits != len(q_indices):
            print(f"[WARNING] Gate '{name}' expects {gate.num_qubits} qubits but got {len(q_indices)} - skipped.")
            continue

        if is_cond and c_indices:
            ctrl_bit = c_indices[0]
            if ctrl_bit >= n_bits:
                print(f"[WARNING] Conditional gate '{name}' references out-of-range classical bit {ctrl_bit} - applied unconditionally.")
                qc.append(gate, q_indices)
            else:
                with qc.if_test((qc.clbits[ctrl_bit], 1)):
                    qc.append(gate, q_indices)
        else:
            qc.append(gate, q_indices)

    return qc

hdh.converters.qiskit_converter.partitions_to_qiskit(hdh, partitions)

Split a partitioned HDH into one Qiskit circuit per partition.

Each circuit keeps only the gates entirely inside its partition; gates across partitions are the cuts a distributed run must implement with communication primitives.

Parameters:
  • partitions (List[Set[str]]) –

    Node sets, e.g. from compute_cut.

Examples:

>>> from qiskit import QuantumCircuit
>>> from hdh.passes.cut import compute_cut
>>> qc = QuantumCircuit(4)
>>> _ = qc.cx(0, 1); _ = qc.cx(2, 3)
>>> hdh = from_qiskit(qc)
>>> partitions, cut_cost = compute_cut(hdh, k=2, cap=2)
>>> cut_cost, [c.num_qubits for c in partitions_to_qiskit(hdh, partitions)]
(0, [2, 2])
Source code in hdh\converters\qiskit_converter.py
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
def partitions_to_qiskit(hdh: HDH, partitions: List[Set[str]]) -> List[QuantumCircuit]:
    """Split a partitioned HDH into one Qiskit circuit per partition.

    Each circuit keeps only the gates entirely inside its partition; gates
    across partitions are the cuts a distributed run must implement with
    communication primitives.

    Args:
        partitions: Node sets, e.g. from `compute_cut`.

    Examples:
        >>> from qiskit import QuantumCircuit
        >>> from hdh.passes.cut import compute_cut
        >>> qc = QuantumCircuit(4)
        >>> _ = qc.cx(0, 1); _ = qc.cx(2, 3)
        >>> hdh = from_qiskit(qc)
        >>> partitions, cut_cost = compute_cut(hdh, k=2, cap=2)
        >>> cut_cost, [c.num_qubits for c in partitions_to_qiskit(hdh, partitions)]
        (0, [2, 2])
    """
    return [to_qiskit(_project_hdh(hdh, node_set)) for node_set in partitions]

hdh.converters.qasm_converter.from_qasm(input_type, qasm)

Convert OpenQASM 2.0 to an HDH, via Qiskit's parser and from_qiskit.

Parameters:
  • input_type (str) –

    "file" if qasm is a path, "string" if it is QASM source.

  • qasm (str) –

    The path or source.

Raises:
  • ValueError –

    If input_type is neither.

Examples:

>>> src = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[2]; h q[0]; cx q[0], q[1];'
>>> from_qasm("string", src).get_num_qubits()
2
Source code in hdh\converters\qasm_converter.py
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def from_qasm(input_type: str, qasm: str):
    """Convert OpenQASM 2.0 to an HDH, via Qiskit's parser and `from_qiskit`.

    Args:
        input_type: ``"file"`` if `qasm` is a path, ``"string"`` if it is
            QASM source.
        qasm: The path or source.

    Raises:
        ValueError: If `input_type` is neither.

    Examples:
        >>> src = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[2]; h q[0]; cx q[0], q[1];'
        >>> from_qasm("string", src).get_num_qubits()
        2
    """
    if input_type == 'file':
        circuit = QuantumCircuit.from_qasm_file(qasm)
    elif input_type == 'string':
        circuit = QuantumCircuit.from_qasm_str(qasm)
    else:
        raise ValueError("Unsupported type. Use 'file' or 'string'.")

    return from_qiskit(circuit)

hdh.converters.cirq_converter.from_cirq(c)

Convert a Cirq circuit to an HDH.

Supports standard gates and measurement. Classically controlled operations are not supported.

Examples:

>>> import cirq
>>> a, b = cirq.LineQubit.range(2)
>>> circuit = cirq.Circuit([cirq.H(a), cirq.CNOT(a, b), cirq.measure(b)])
>>> from_cirq(circuit).get_num_qubits()
2
Source code in hdh\converters\cirq_converter.py
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def from_cirq(c: cirq.Circuit) -> HDH:
    """Convert a Cirq circuit to an HDH.

    Supports standard gates and measurement. Classically controlled
    operations are not supported.

    Examples:
        >>> import cirq
        >>> a, b = cirq.LineQubit.range(2)
        >>> circuit = cirq.Circuit([cirq.H(a), cirq.CNOT(a, b), cirq.measure(b)])
        >>> from_cirq(circuit).get_num_qubits()
        2
    """
    circuit = Circuit()
    qmap = _qubit_index_map(c)

    for moment in c:
        for op in moment.operations:
            if _is_measure(op):
                q_indices = [qmap[q] for q in op.qubits]
                circuit.add_instruction("measure", q_indices, None)
                continue

            name = _gate_name(op.gate)
            q_indices = [qmap[q] for q in op.qubits]
            modifies_flags = [True] * len(q_indices)
            circuit.add_instruction(name, q_indices, None, modifies_flags)

    return circuit.build_hdh()

hdh.converters.pennylane_converter.from_pennylane(circ_like)

Convert a PennyLane QuantumScript (or recorded operations) to an HDH.

Supports standard gates, mid-circuit and terminal measurements, and qml.cond on a single measurement. Wire labels are renumbered 0..n-1 in declaration order.

Raises:
  • NotImplementedError –

    If a qml.cond condition is not a measurement.

Examples:

>>> import pennylane as qml
>>> script = qml.tape.QuantumScript(
...     [qml.Hadamard(0), qml.CNOT([0, 1])], [qml.probs(wires=1)])
>>> from_pennylane(script).get_num_qubits()
2
Source code in hdh\converters\pennylane_converter.py
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def from_pennylane(circ_like: Union[QuantumScript, OperationRecorder]) -> HDH:
    """Convert a PennyLane `QuantumScript` (or recorded operations) to an HDH.

    Supports standard gates, mid-circuit and terminal measurements, and
    `qml.cond` on a single measurement. Wire labels are renumbered
    ``0..n-1`` in declaration order.

    Raises:
        NotImplementedError: If a `qml.cond` condition is not a measurement.

    Examples:
        >>> import pennylane as qml
        >>> script = qml.tape.QuantumScript(
        ...     [qml.Hadamard(0), qml.CNOT([0, 1])], [qml.probs(wires=1)])
        >>> from_pennylane(script).get_num_qubits()
        2
    """
    qs = circ_like
    wire2idx = _wire_index_map(qs)

    circuit = Circuit()

    # Track which mid-measure maps to which classical bit index
    meas_mp_to_cbit: Dict[MidMeasureMP, int] = {}
    next_cbit = 0

    for op in qs.operations:
        # mid-circuit measure -> explicit "measure" instruction
        if isinstance(op, (MidMeasureMP,ProbabilityMP, ExpectationMP, SampleMP)):
            w = op.wires[0]
            cbit = meas_mp_to_cbit.setdefault(op, next_cbit)
            if cbit == next_cbit:
                next_cbit += 1
            circuit.add_instruction("measure", [wire2idx[w]], [cbit])
            continue

        # conditional op (qml.cond -> Conditional container)
        if isinstance(op, Conditional):
            then = op.base
            mval = op.meas_val
            mps = getattr(mval, "measurements", None)
            if not mps:
                raise NotImplementedError("Only MeasurementValue-based conditions are supported.")
            mp = mps[0]  # single-bit condition
            cbit = meas_mp_to_cbit.setdefault(mp, next_cbit)
            if cbit == next_cbit:
                next_cbit += 1

            qidxs = [wire2idx[w] for w in then.wires]
            then_params = [float(p) for p in then.parameters] if then.parameters else None
            circuit.add_instruction(
                then.name.lower(),
                qidxs,
                bits=[cbit],
                modifies_flags=[True] * len(qidxs),
                cond_flag="p",
                params=then_params,
            )
            continue

        # plain operation
        name = op.name.lower()
        qidxs = [wire2idx[w] for w in op.wires]
        params = [float(p) for p in op.parameters] if op.parameters else None
        circuit.add_instruction(name, qidxs, bits=[], params=params)

    return circuit.build_hdh()

hdh.converters.pennylane_converter.to_pennylane(hdh)

Convert an HDH back into a PennyLane QuantumScript.

Each multi-qubit gate is rebuilt from its _stage2 hyperedge, as in to_qiskit. Measurements and conditional gates are replayed through qml.measure and qml.cond.

Examples:

>>> import pennylane as qml
>>> script = qml.tape.QuantumScript([qml.Hadamard(0), qml.CNOT([0, 1])])
>>> [op.name for op in to_pennylane(from_pennylane(script)).operations]
['Hadamard', 'CNOT']
Source code in hdh\converters\pennylane_converter.py
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def to_pennylane(hdh: HDH) -> QuantumScript:
    """Convert an HDH back into a PennyLane `QuantumScript`.

    Each multi-qubit gate is rebuilt from its ``_stage2`` hyperedge, as in
    `to_qiskit`. Measurements and conditional gates are replayed through
    `qml.measure` and `qml.cond`.

    Examples:
        >>> import pennylane as qml
        >>> script = qml.tape.QuantumScript([qml.Hadamard(0), qml.CNOT([0, 1])])
        >>> [op.name for op in to_pennylane(from_pennylane(script)).operations]
        ['Hadamard', 'CNOT']
    """
    qubit_indices: Set[int] = set()
    for node_id in hdh.nodes:
        q = _parse_qubit(node_id)
        if q is not None:
            qubit_indices.add(q)

    qubit_map = {q: i for i, q in enumerate(sorted(qubit_indices))}

    if not qubit_map:
        return qml.tape.make_qscript(lambda: None)()

    # (sort_time, name, wires, cbit, is_cond, params)
    records: List[tuple] = []

    for edge in hdh.hyperedges:
        raw_name = hdh.gate_name.get(edge, "")
        edge_type = hdh.hyperedge_types.get(edge, "q")

        if raw_name == "measure":
            q_nodes = sorted(
                (n for n in edge if hdh.node_types.get(n) == "q"),
                key=lambda n: hdh.time_map.get(n, 0),
            )
            c_nodes = sorted(
                (n for n in edge if hdh.node_types.get(n) == "c"),
                key=lambda n: hdh.time_map.get(n, 0),
            )
            if not q_nodes or not c_nodes:
                continue
            q_idx = _parse_qubit(q_nodes[0])
            c_idx = _parse_cbit(c_nodes[0])
            if q_idx is None or c_idx is None or q_idx not in qubit_map:
                continue
            sort_time = hdh.time_map.get(c_nodes[0], 0)
            records.append((sort_time, "measure", [qubit_map[q_idx]], c_idx, False, None))
            continue

        # Stages 1 and 3 only carry wire continuity; stage 2 holds the real gate.
        if raw_name.endswith("_stage1") or raw_name.endswith("_stage3"):
            continue

        # Classical edges of a conditional gate duplicate its quantum edge.
        if edge_type == "c":
            continue

        args = hdh.edge_args.get(edge)
        if args is None:
            continue

        q_with_time, c_with_time, _ = args
        actual_name = raw_name[:-7] if raw_name.endswith("_stage2") else raw_name
        wires = [qubit_map[q] for q, _ in q_with_time if q in qubit_map]
        sort_time = min((t for _, t in q_with_time), default=0)
        is_cond = hdh.hyperedge_realisation.get(edge) == "p"
        cbit = c_with_time[0][0] if (is_cond and c_with_time) else None

        records.append((sort_time, actual_name, wires, cbit, is_cond, hdh.gate_params.get(edge)))

    records.sort(key=lambda r: r[0])

    # A multi-qubit gate contributes one record per qubit wire; keep one.
    seen: Set[tuple] = set()
    unique_records: List[tuple] = []
    for rec in records:
        sort_time, name, wires, cbit, is_cond, params = rec
        dedup_key = (sort_time, name, tuple(wires))
        if dedup_key not in seen:
            seen.add(dedup_key)
            unique_records.append(rec)

    def _replay():
        cbit_to_mp: Dict[int, Any] = {}
        for sort_time, name, wires, cbit, is_cond, params in unique_records:
            if name == "measure":
                cbit_to_mp[cbit] = qml.measure(wires[0])
                continue

            if is_cond and cbit in cbit_to_mp:
                qml.cond(cbit_to_mp[cbit], lambda n=name, p=params, w=wires: _mk_op(n, p, w))()
            else:
                if is_cond:
                    warnings.warn(
                        f"[to_pennylane] Conditional gate '{name}' at t={sort_time} has no "
                        f"matching measurement - applying unconditionally."
                    )
                _mk_op(name, params, wires)

    return qml.tape.make_qscript(_replay)()

hdh.converters.braket_converter.from_braket(bk)

Convert an Amazon Braket Circuit to an HDH.

Gates, including noise operations, are mapped by name, and measurements are supported. Classical control flow raises NotImplementedError.

Examples:

>>> from braket.circuits import Circuit as BraketCircuit
>>> from_braket(BraketCircuit().h(0).cnot(0, 1)).get_num_qubits()
2
Source code in hdh\converters\braket_converter.py
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def from_braket(bk: BraketCircuit) -> HDH:
    """Convert an Amazon Braket `Circuit` to an HDH.

    Gates, including noise operations, are mapped by name, and measurements
    are supported. Classical control flow raises `NotImplementedError`.

    Examples:
        >>> from braket.circuits import Circuit as BraketCircuit
        >>> from_braket(BraketCircuit().h(0).cnot(0, 1)).get_num_qubits()
        2
    """
    qmap = _qindex_map(bk)
    circuit = Circuit()

    for instr in bk.instructions:
        if _is_ignorable(instr):
            continue

        name = _name_of(instr)
        q_indices = [qmap[q] for q in instr.target]

        # Basic check for classical control (Braket may wrap ops with conditions in newer APIs)
        if hasattr(instr, "condition") and instr.condition is not None:
            # Only implement if you have a bit-index mapping. Braket conditions are not 1-bit Clbits.
            raise NotImplementedError("Classical conditionals in Braket are not supported yet.")

        if _is_measure(instr):
            # Measurement is per qubit target
            circuit.add_instruction("measure", q_indices, None)
            continue

        # Treat any other op (gate or noise) as a modifying quantum op
        modifies_flags = [True] * len(q_indices)
        circuit.add_instruction(name, q_indices, [], modifies_flags)

    return circuit.build_hdh()

Partitioning

hdh.passes.cut.compute_cut(hdh_graph, k, cap, *, beam_k=3, backtrack_window=0, polish_1swap_budget=0, restarts=1, reserve_frac=0.08, predictive_reject=True, seed=0)

Capacity-aware greedy partitioner: split an HDH across k devices of cap qubits each, minimising cut hyperedges.

It assigns nodes (timestepped states) rather than whole qubits, so a qubit's history may be split between devices when that is cheaper. Capacity counts distinct quantum wires per device, and a split qubit uses a slot on each device it touches, so splits are only made while enough slots remain for every qubit not yet placed. This guarantees a complete assignment whenever k * cap >= n_qubits. The remaining keyword arguments are accepted for backwards compatibility and ignored.

Parameters:
  • hdh_graph –

    The HDH to partition.

  • k (int) –

    Number of devices.

  • cap (int) –

    Qubits per device.

  • beam_k (int, default: 3 ) –

    How many of the earliest frontier nodes to score per step.

Returns:
  • List[Set[str]] –

    (partitions, cost): k sets of node IDs, and the number of

  • int –

    hyperedges spanning more than one partition.

Raises:
  • RuntimeError –

    If no complete assignment exists, i.e. k * cap is smaller than the number of qubits.

Examples:

>>> from hdh.models.circuit import Circuit
>>> c = Circuit()
>>> for q in range(3):
...     c.add_instruction("cx", [q, q + 1])
>>> partitions, cost = compute_cut(c.build_hdh(), k=2, cap=2)
>>> cost
1
Source code in hdh\passes\cut.py
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
def compute_cut(hdh_graph, k: int, cap: int, *,
                beam_k: int = 3,
                backtrack_window: int = 0,
                polish_1swap_budget: int = 0,
                restarts: int = 1,
                reserve_frac: float = 0.08,
                predictive_reject: bool = True,
                seed: int = 0) -> Tuple[List[Set[str]], int]:
    """Capacity-aware greedy partitioner: split an HDH across `k` devices of
    `cap` qubits each, minimising cut hyperedges.

    It assigns nodes (timestepped states) rather than whole qubits, so a
    qubit's history may be split between devices when that is cheaper.
    Capacity counts distinct quantum wires per device, and a split qubit uses
    a slot on each device it touches, so splits are only made while enough
    slots remain for every qubit not yet placed. This guarantees a complete
    assignment whenever ``k * cap >= n_qubits``. The remaining keyword
    arguments are accepted for backwards compatibility and ignored.

    Args:
        hdh_graph: The HDH to partition.
        k: Number of devices.
        cap: Qubits per device.
        beam_k: How many of the earliest frontier nodes to score per step.

    Returns:
        ``(partitions, cost)``: `k` sets of node IDs, and the number of
        hyperedges spanning more than one partition.

    Raises:
        RuntimeError: If no complete assignment exists, i.e. ``k * cap`` is
            smaller than the number of qubits.

    Examples:
        >>> from hdh.models.circuit import Circuit
        >>> c = Circuit()
        >>> for q in range(3):
        ...     c.add_instruction("cx", [q, q + 1])
        >>> partitions, cost = compute_cut(c.build_hdh(), k=2, cap=2)
        >>> cost
        1
    """
    if not hdh_graph.nodes or not hdh_graph.hyperedges:
        return [set() for _ in range(k)], 0

    # Build temporal incidence structure
    inc, pins = _build_temporal_incidence(hdh_graph)

    # Initialize partitions and tracking structures
    partitions = [set() for _ in range(k)]
    unassigned = set(hdh_graph.nodes)
    partition_qubits = [set() for _ in range(k)]  # Track unique qubits per partition
    used = [0] * k  # Track number of unique qubits used per partition

    # Capacity is charged per (bin, qubit) pair, so a qubit whose timeline is
    # split across bins occupies a slot in each one. That makes splits a scarce
    # resource: of the k*cap slots, one is needed per still-unplaced qubit and
    # only the surplus can be spent on splitting. Tracking that surplus is what
    # keeps the greedy from splitting itself into a state where some qubit has
    # nowhere left to go.
    all_qubits = {q for q in (_qubit_of(hdh_graph, n) for n in hdh_graph.nodes)
                  if q is not None}
    assigned_qubits: Set[str] = set()   # qubits held by at least one bin
    total_slots = k * cap
    slots_used = 0                      # == sum(len(pq) for pq in partition_qubits)

    def _split_is_safe() -> bool:
        """Whether one more split still leaves every unplaced qubit a slot."""
        return (total_slots - slots_used - 1) >= (len(all_qubits) - len(assigned_qubits))

    def _place(node: str, bin_idx: int) -> None:
        """Assign a node to a bin, keeping qubit and slot bookkeeping in step."""
        nonlocal slots_used
        partitions[bin_idx].add(node)
        unassigned.discard(node)
        q = _qubit_of(hdh_graph, node)
        if q is None:
            return
        if q not in partition_qubits[bin_idx]:
            partition_qubits[bin_idx].add(q)
            used[bin_idx] = len(partition_qubits[bin_idx])
            slots_used += 1
        assigned_qubits.add(q)

    # QPU order (for now, just sequential; could be topology-aware)
    qpu_order = list(range(k))

    # Phase 1 & 2: Greedy bin filling with cost-aware frontier selection
    for i in range(k):
        bin_idx = qpu_order[i]

        if not unassigned:
            break

        # Seed on a qubit no bin owns yet. Seeding on an already-placed qubit
        # opens the bin with a split, which is what used to strand later qubits
        # when capacity was tight. If only such nodes remain they cost nothing
        # to absorb into the bin that already holds them, so leave them to the
        # residual phase.
        seed_pool = [n for n in unassigned
                     if _qubit_of(hdh_graph, n) not in assigned_qubits]
        if not seed_pool:
            break
        seed_node = min(seed_pool, key=lambda n: (hdh_graph.time_map.get(n, 0), n))

        # Initialize bin with seed - update partitions immediately
        _place(seed_node, bin_idx)

        # Initialize frontier with seed's neighbors
        frontier = []  # Min-heap of (time, counter, node)
        counter = [0]  # Counter for tie-breaking
        rejected = set()  # Nodes rejected due to capacity in this bin
        _push_next_valid_neighbors(hdh_graph, seed_node, frontier, unassigned, inc, pins, counter)

        # Greedy temporal expansion with cost-aware selection. Deliberately not
        # bounded by `used[bin_idx] < cap`: a bin holding `cap` qubits can still
        # absorb further nodes of those same qubits at zero capacity cost, and
        # stopping at the cap left exactly those nodes behind to be picked up as
        # the next bin's seed - splitting a qubit that never needed splitting.
        while True:
            # Select best node from frontier using delta cost (excluding rejected)
            next_node = _select_best_from_frontier_with_rejected(frontier, unassigned, rejected,
                                                                  bin_idx, partitions, inc, pins, beam_k,
                                                                  partition_qubits, hdh_graph)

            if next_node is None:
                break  # No more valid neighbors

            # Check if adding this node would exceed capacity
            next_qubit = _qubit_of(hdh_graph, next_node)
            if next_qubit is not None and next_qubit not in partition_qubits[bin_idx]:
                if used[bin_idx] + 1 > cap:
                    # Would exceed capacity, reject and try next candidate
                    rejected.add(next_node)
                    continue
                if next_qubit in assigned_qubits and not _split_is_safe():
                    # This split would claim the last slot some unplaced qubit needs
                    rejected.add(next_node)
                    continue

            # Add node to partition immediately (not just to local variable)
            _place(next_node, bin_idx)

            # Push this node's neighbors to frontier
            _push_next_valid_neighbors(hdh_graph, next_node, frontier, unassigned, inc, pins, counter)

    # Phase 3: Residual best-fit placement with delta cost
    unplaceable_nodes = set()

    while unassigned:
        # Find the next unassigned node
        remaining = unassigned - unplaceable_nodes
        if not remaining:
            break  # All remaining nodes are unplaceable

        node = min(remaining, key=lambda n: (hdh_graph.time_map.get(n, 0), n))
        node_qubit = _qubit_of(hdh_graph, node)

        # Compute delta cost for each bin and find best fit. Among bins with
        # equal delta, prefer one that already owns the qubit: that placement is
        # free, where any other bin spends a capacity slot on a split.
        best_bin = None
        best_score = None

        for bin_idx in range(k):
            # Check capacity constraint
            is_split = False
            if node_qubit is not None and node_qubit not in partition_qubits[bin_idx]:
                if used[bin_idx] >= cap:
                    continue
                is_split = node_qubit in assigned_qubits
                if is_split and not _split_is_safe():
                    continue

            delta = _compute_delta_cost_simple(node, bin_idx, partitions, inc, pins)
            score = (delta, 1 if is_split else 0, bin_idx)
            if best_score is None or score < best_score:
                best_score = score
                best_bin = bin_idx

        # Place node in best bin
        if best_bin is not None:
            _place(node, best_bin)
        else:
            # Try teledata fallback: find a bin where this qubit already exists
            placed = False
            if node_qubit is not None:
                for b in range(k):
                    if node_qubit in partition_qubits[b]:
                        _place(node, b)
                        placed = True
                        break

            if not placed:
                # Node is truly unplaceable, mark it and try next node
                unplaceable_nodes.add(node)

    if unplaceable_nodes:
        raise RuntimeError(
            f"compute_cut: the greedy heuristic could not place {len(unplaceable_nodes)} "
            f"node(s) within the given capacity constraints (k={k}, cap={cap}). Returning "
            f"a partial partition would silently under-report the cut cost, so this is "
            f"raised instead. Try a larger `cap` and/or `k`."
        )

    # Compute cost (count cut hyperedges)
    node_assignment = {}
    for partition_idx, partition_nodes in enumerate(partitions):
        for node in partition_nodes:
            node_assignment[node] = partition_idx

    cost = _compute_cut_cost(hdh_graph, node_assignment)

    return partitions, cost

hdh.passes.cut.kahypar_cutter(hdh, k, cap, *, seed=0, config_path=None, suppress_output=True)

Partition by qubit with KaHyPar, keeping each qubit on one device.

Each quantum wire is one KaHyPar vertex, and the balance tolerance is set so no block exceeds cap qubits. Classical nodes go with a qubit they share a hyperedge with. Requires the kahypar extra.

Parameters:
  • k (int) –

    Number of devices.

  • cap (int) –

    Qubits per device.

  • seed (int, default: 0 ) –

    Passed to KaHyPar when supported.

  • config_path (Optional[str], default: None ) –

    KaHyPar INI file; defaults to the bundled one.

  • suppress_output (bool, default: True ) –

    Silence KaHyPar's output.

Returns:
  • –

    (partitions, cut_cost), as in compute_cut.

Raises:
  • ImportError –

    If kahypar is not installed.

  • ValueError –

    If k exceeds the number of qubits, or cap is below ceil(n_qubits / k).

Examples:

>>> partitions, cut_cost = kahypar_cutter(hdh, k=2, cap=3)
Source code in hdh\passes\cut.py
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def kahypar_cutter(
    hdh,
    k: int,
    cap: int,
    *,
    seed: int = 0,
    config_path: Optional[str] = None,
    suppress_output: bool = True,
):
    """Partition by qubit with KaHyPar, keeping each qubit on one device.

    Each quantum wire is one KaHyPar vertex, and the balance tolerance is set
    so no block exceeds `cap` qubits. Classical nodes go with a qubit they
    share a hyperedge with. Requires the ``kahypar`` extra.

    Args:
        k: Number of devices.
        cap: Qubits per device.
        seed: Passed to KaHyPar when supported.
        config_path: KaHyPar INI file; defaults to the bundled one.
        suppress_output: Silence KaHyPar's output.

    Returns:
        ``(partitions, cut_cost)``, as in `compute_cut`.

    Raises:
        ImportError: If ``kahypar`` is not installed.
        ValueError: If `k` exceeds the number of qubits, or `cap` is below
            ``ceil(n_qubits / k)``.

    Examples:
        >>> partitions, cut_cost = kahypar_cutter(hdh, k=2, cap=3)  # doctest: +SKIP
    """
    try:
        import kahypar  # type: ignore
    except Exception as e:
        raise ImportError(
            "`kahypar` is not installed. Install it (e.g. `pip install kahypar==1.3.6`) "
            "and ensure you have the build requirements from the PyPI page."
        ) from e

    if k <= 0:
        raise ValueError("k must be >= 1")

    # --- Build qubit vertex set ---
    qubits: Set[str] = set()
    qubit_nodes = defaultdict(list)  # q -> [node ids]
    classical_nodes: List[str] = []

    for nid in hdh.nodes:
        q = _qubit_of(hdh, nid)
        if q is not None:
            qubits.add(q)
            qubit_nodes[q].append(nid)
        else:
            classical_nodes.append(nid)

    qubit_list = sorted(qubits)
    n = len(qubit_list)
    if n == 0:
        # No qubits: nothing meaningful to partition.
        return [set() for _ in range(k)], 0

    if k > n:
        raise ValueError(f"k={k} cannot exceed number of qubits n={n} for kahypar_cutter")

    if cap <= 0:
        raise ValueError("cap must be >= 1 (capacity = max unique qubits per partition)")

    # If cap is too small to ever fit an even split, fail early.
    min_required = (n + k - 1) // k
    if cap < min_required:
        raise ValueError(
            f"cap={cap} is too small for n={n}, k={k}. "
            f"Need at least ceil(n/k)={min_required} to be feasible."
        )

    q_to_vid = {q: i for i, q in enumerate(qubit_list)}

    # --- Build undirected hyperedges on qubit-vertices ---
    # KaHyPar expects hyperedges as pins (vertex IDs). We collapse each HDH hyperedge
    # to the set of qubits it touches.
    hedge_pins: List[List[int]] = []
    hedge_weights: List[int] = []

    for e in hdh.hyperedges:
        qs = {_qubit_of(hdh, nid) for nid in e} - {None}
        if len(qs) >= 2:
            hedge_pins.append([q_to_vid[q] for q in sorted(qs)])
            hedge_weights.append(1)

    # Degenerate case: no multi-qubit couplings
    if not hedge_pins:
        # Purely disconnected qubits: just pack sequentially.
        partitions = [set() for _ in range(k)]
        qubit_block = {q: i % k for i, q in enumerate(qubit_list)}
        for q, b in qubit_block.items():
            partitions[b].update(qubit_nodes[q])
        for nid, b in _attach_classical(hdh, classical_nodes, qubit_block).items():
            partitions[b].add(nid)
        return partitions, _compute_cut_cost(hdh, {nid: b for b, part in enumerate(partitions) for nid in part})

    # Convert pins list to CSR-style arrays required by KaHyPar.
    # hyperedge_indices: prefix-sum offsets into `hyperedges` array.
    hyperedge_indices = [0]
    hyperedges: List[int] = []
    for pins in hedge_pins:
        hyperedges.extend(pins)
        hyperedge_indices.append(len(hyperedges))

    num_hyperedges = len(hedge_pins)
    vertex_weights = [1] * n  # each qubit counts as 1 capacity unit

    # --- KaHyPar context/config ---
    if config_path is None:
        config_path = _DEFAULT_KAHYPAR_CONFIG
        if not os.path.exists(config_path):
            raise FileNotFoundError(
                "KaHyPar needs an INI configuration file, and the bundled default "
                f"({_DEFAULT_KAHYPAR_CONFIG}) is missing. "
                "Pass `config_path=...` (e.g. a KaHyPar .ini file)."
            )

    # Balance: ensure max block size <= cap.
    target = n / float(k)
    epsilon = max(0.0, (cap / target) - 1.0)

    context = kahypar.Context()
    context.loadINIconfiguration(str(config_path))
    context.setK(k)
    context.setEpsilon(epsilon)

    # Some builds support setting a seed.
    if hasattr(context, "setSeed"):
        try:
            context.setSeed(int(seed))
        except Exception:
            pass

    if suppress_output and hasattr(context, "suppressOutput"):
        try:
            context.suppressOutput(True)
        except Exception:
            pass

    # --- Build and partition hypergraph ---
    # KaHyPar signature differs slightly across versions; try the common one.
    try:
        hg = kahypar.Hypergraph(
            n,
            num_hyperedges,
            hyperedge_indices,
            hyperedges,
            hedge_weights,
            vertex_weights,
        )
    except TypeError:
        # Older signature: Hypergraph(num_vertices, num_hyperedges, hyperedge_indices, hyperedges, k, ...)
        hg = kahypar.Hypergraph(
            n,
            num_hyperedges,
            hyperedge_indices,
            hyperedges,
            k,
            hedge_weights,
            vertex_weights,
        )

    kahypar.partition(hg, context)

    # --- Read partitioning and lift back to HDH nodes ---
    qubit_block = {qubit_list[v]: int(hg.blockID(v)) for v in range(n)}

    partitions: List[Set[str]] = [set() for _ in range(k)]
    for q, nodes in qubit_nodes.items():
        partitions[qubit_block[q]].update(nodes)
    for nid, b in _attach_classical(hdh, classical_nodes, qubit_block).items():
        partitions[b].add(nid)

    node_assignment = {nid: b for b, part in enumerate(partitions) for nid in part}
    cut_cost = _compute_cut_cost(hdh, node_assignment)
    return partitions, cut_cost

hdh.passes.cut.kahypar_cutter_nodebalanced(hdh, k, *, seed=0, config_path=None, epsilon=0.03, suppress_output=True)

Partition nodes with KaHyPar, balancing node counts and ignoring qubit capacity.

A capacity-oblivious baseline: every node is a vertex of weight 1, and blocks are balanced within epsilon in node count, not qubits. Useful for measuring how often such a partitioner violates qubit capacity. Requires the kahypar extra.

Returns:
  • –

    (partitions, cut_cost), as in compute_cut.

Source code in hdh\passes\cut.py
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
def kahypar_cutter_nodebalanced(
    hdh,
    k: int,
    *,
    seed: int = 0,
    config_path: Optional[str] = None,
    epsilon: float = 0.03,
    suppress_output: bool = True,
):
    """Partition nodes with KaHyPar, balancing node counts and ignoring qubit
    capacity.

    A capacity-oblivious baseline: every node is a vertex of weight 1, and
    blocks are balanced within `epsilon` in node count, not qubits. Useful for
    measuring how often such a partitioner violates qubit capacity. Requires
    the ``kahypar`` extra.

    Returns:
        ``(partitions, cut_cost)``, as in `compute_cut`.
    """
    try:
        import kahypar  # type: ignore
    except Exception as e:
        raise ImportError(
            "`kahypar` is not installed. Install it (e.g. `pip install kahypar==1.3.6`)."
        ) from e

    if k <= 0:
        raise ValueError("k must be >= 1")

    nodes = sorted(list(hdh.nodes))
    n = len(nodes)
    if n == 0:
        return [set() for _ in range(k)], 0
    if k > n:
        raise ValueError(f"k={k} cannot exceed number of HDH nodes n={n} for node-balanced KaHyPar")

    # Build node id -> vertex id
    nid_to_vid = {nid: i for i, nid in enumerate(nodes)}

    # Build undirected hyperedges as pins (vertex IDs)
    hedge_pins: List[List[int]] = []
    hedge_weights: List[int] = []
    for e in hdh.hyperedges:
        pins = [nid_to_vid[nid] for nid in e if nid in nid_to_vid]
        pins = sorted(set(pins))
        if len(pins) >= 2:
            hedge_pins.append(pins)
            hedge_weights.append(1)

    # Degenerate case: no usable hyperedges
    if not hedge_pins:
        parts = [set() for _ in range(k)]
        for i, nid in enumerate(nodes):
            parts[i % k].add(nid)
        node_assignment = {nid: b for b, part in enumerate(parts) for nid in part}
        return parts, _compute_cut_cost(hdh, node_assignment)

    hyperedge_indices = [0]
    hyperedges: List[int] = []
    for pins in hedge_pins:
        hyperedges.extend(pins)
        hyperedge_indices.append(len(hyperedges))

    num_hyperedges = len(hedge_pins)
    vertex_weights = [1] * n

    if config_path is None:
        config_path = _DEFAULT_KAHYPAR_CONFIG
        if not os.path.exists(config_path):
            raise FileNotFoundError(
                "KaHyPar needs an INI configuration file, and the bundled default "
                f"({_DEFAULT_KAHYPAR_CONFIG}) is missing. "
                "Pass `config_path=...` (e.g. a KaHyPar .ini file)."
            )

    context = kahypar.Context()
    context.loadINIconfiguration(str(config_path))
    context.setK(k)
    context.setEpsilon(float(epsilon))

    if hasattr(context, "setSeed"):
        try:
            context.setSeed(int(seed))
        except Exception:
            pass
    if suppress_output and hasattr(context, "suppressOutput"):
        try:
            context.suppressOutput(True)
        except Exception:
            pass

    try:
        hg = kahypar.Hypergraph(
            n,
            num_hyperedges,
            hyperedge_indices,
            hyperedges,
            hedge_weights,
            vertex_weights,
        )
    except TypeError:
        hg = kahypar.Hypergraph(
            n,
            num_hyperedges,
            hyperedge_indices,
            hyperedges,
            k,
            hedge_weights,
            vertex_weights,
        )

    kahypar.partition(hg, context)

    partitions: List[Set[str]] = [set() for _ in range(k)]
    for vid, nid in enumerate(nodes):
        b = int(hg.blockID(vid))
        partitions[b].add(nid)

    node_assignment = {nid: b for b, part in enumerate(partitions) for nid in part}
    cut_cost = _compute_cut_cost(hdh, node_assignment)
    return partitions, cut_cost

hdh.passes.cut.metis_telegate(hdh, partitions, capacities)

Partition qubits with METIS on the telegate graph, then repair bins over capacity.

The telegate graph has one vertex per quantum wire and edges weighted by how many quantum operations two wires share. Falls back to Kernighan-Lin bisection when METIS is unavailable.

Parameters:
  • partitions (int) –

    Number of bins.

  • capacities (int) –

    Qubits per bin.

Returns:
  • List[Set[str]] –

    (bins, cut_cost, respects_capacity, method): sets of wire labels,

  • int –

    the number of cut graph edges, whether every bin fits, and

  • bool –

    "metis" or "kl".

Examples:

>>> from hdh.models.circuit import Circuit
>>> c = Circuit()
>>> c.add_instruction("cx", [0, 1])
>>> c.add_instruction("cx", [2, 3])
>>> bins, cut_cost, fits, method = metis_telegate(c.build_hdh(), 2, 2)
>>> sorted(sorted(b) for b in bins), cut_cost, fits
([['q0', 'q1'], ['q2', 'q3']], 0, True)
Source code in hdh\passes\cut.py
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
def metis_telegate(hdh: "HDH", partitions: int, capacities: int) -> Tuple[List[Set[str]], int, bool, str]:
    """Partition qubits with METIS on the telegate graph, then repair bins over
    capacity.

    The telegate graph has one vertex per quantum wire and edges weighted by
    how many quantum operations two wires share. Falls back to
    Kernighan-Lin bisection when METIS is unavailable.

    Args:
        partitions: Number of bins.
        capacities: Qubits per bin.

    Returns:
        ``(bins, cut_cost, respects_capacity, method)``: sets of wire labels,
        the number of cut graph edges, whether every bin fits, and
        ``"metis"`` or ``"kl"``.

    Examples:
        >>> from hdh.models.circuit import Circuit
        >>> c = Circuit()
        >>> c.add_instruction("cx", [0, 1])
        >>> c.add_instruction("cx", [2, 3])
        >>> bins, cut_cost, fits, method = metis_telegate(c.build_hdh(), 2, 2)
        >>> sorted(sorted(b) for b in bins), cut_cost, fits
        ([['q0', 'q1'], ['q2', 'q3']], 0, True)
    """
    G: nx.Graph = telegate_hdh(hdh)

    used_metis = False
    try:
        import metis  # type: ignore
        node_order = list(G.nodes())
        _, part_ids = metis.part_graph(G, nparts=partitions)
        qubit_parts = [set() for _ in range(partitions)]
        for node, p in zip(node_order, part_ids):
            qubit_parts[p].add(node)
        used_metis = True
    except Exception:
        qubit_parts = _kl_fallback_partition(G, partitions)

    bins = _bins_from_parts(qubit_parts)
    bins = _repair_overflow(G, bins, capacities)

    cut_cost = _cut_edges_unweighted(G, bins)
    sizes = _sizes(bins)
    respects_capacity = all(s <= capacities for s in sizes)
    method = "metis" if used_metis else "kl"
    return bins, cut_cost, respects_capacity, method

hdh.passes.cut.cost(hdh_graph, partitions)

Count cut hyperedges, separately for quantum and classical ones.

A hyperedge is cut when its nodes lie in more than one partition; nodes in no partition are ignored.

Returns:
  • Tuple[float, float] –

    (quantum_cuts, classical_cuts), as floats.

Examples:

>>> from hdh.models.circuit import Circuit
>>> c = Circuit()
>>> c.add_instruction("cx", [0, 1])
>>> hdh = c.build_hdh()
>>> q0 = {n for n in hdh.nodes if hdh.wire_of[n] == "q0"}
>>> cost(hdh, [q0, hdh.nodes - q0])
(1.0, 0.0)
Source code in hdh\passes\cut.py
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
def cost(hdh_graph, partitions) -> Tuple[float, float]:
    """Count cut hyperedges, separately for quantum and classical ones.

    A hyperedge is cut when its nodes lie in more than one partition; nodes
    in no partition are ignored.

    Returns:
        ``(quantum_cuts, classical_cuts)``, as floats.

    Examples:
        >>> from hdh.models.circuit import Circuit
        >>> c = Circuit()
        >>> c.add_instruction("cx", [0, 1])
        >>> hdh = c.build_hdh()
        >>> q0 = {n for n in hdh.nodes if hdh.wire_of[n] == "q0"}
        >>> cost(hdh, [q0, hdh.nodes - q0])
        (1.0, 0.0)
    """
    if not partitions or not hasattr(hdh_graph, 'hyperedges'):
        return 0.0, 0.0

    # Create mapping from node to partition index
    node_to_partition = {}
    for i, partition in enumerate(partitions):
        for node in partition:
            node_to_partition[node] = i

    # Count hyperedges that cross partitions (separated by type)
    cost_q = 0  # Quantum cost
    cost_c = 0  # Classical cost

    for edge in hdh_graph.hyperedges:
        # Get partitions of all nodes in this hyperedge
        edge_partitions = set()
        for node in edge:
            if node in node_to_partition:
                edge_partitions.add(node_to_partition[node])

        # If hyperedge spans multiple partitions, it contributes to cost
        if len(edge_partitions) > 1:
            # Get edge weight if available
            edge_weight = 1
            if hasattr(hdh_graph, 'edge_weight'):
                edge_weight = hdh_graph.edge_weight.get(edge, 1)

            # Determine if edge is quantum or classical
            edge_type = 'q'  # Default to quantum
            if hasattr(hdh_graph, 'hyperedge_types'):
                edge_type = hdh_graph.hyperedge_types.get(edge, 'q')

            if edge_type == 'q':
                cost_q += edge_weight
            else:
                cost_c += edge_weight

    return float(cost_q), float(cost_c)