Term streaming¶

For large systems, the number of terms in the symbolic history can grow very large. It might become impractical to store all of them in memory at once for post-processing and analysis. In this case, we can stream the terms from disk, which greatly reduces the memory footprint at any given point in time.

Setup¶

We'll use a toy system to demonstrate the streaming functionality.

In [1]:

import numpy as np
from qiskit import QuantumCircuit
from qiskit.circuit.library import (
    XXPlusYYGate,
    PhaseGate,
    RZGate,
    CPhaseGate,
    SwapGate,
    XGate,
)

GATES = [
    (lambda: XXPlusYYGate(np.random.uniform(0, 2 * np.pi), np.random.uniform(0, 2 * np.pi)), 2),
    (lambda: PhaseGate(np.random.uniform(0, 2 * np.pi)), 1),
    (lambda: RZGate(np.random.uniform(0, 2 * np.pi)), 1),
    (lambda: CPhaseGate(np.random.uniform(0, 2 * np.pi)), 2),
    (lambda: SwapGate(), 2),
    (lambda: XGate(), 1),
]

np.random.seed(0)
n_qubits = 8
qc = QuantumCircuit(n_qubits)

for _ in range(120):
    factory, nq = GATES[np.random.randint(len(GATES))]
    gate = factory()
    qubits = np.random.choice(n_qubits, size=nq, replace=False).tolist()
    qc.append(gate, qubits)

from qiskit.quantum_info import SparsePauliOp

observable = SparsePauliOp.from_list([("X" + "I" * (n_qubits - 1), 1.0)])

import numpy as np from qiskit import QuantumCircuit from qiskit.circuit.library import ( XXPlusYYGate, PhaseGate, RZGate, CPhaseGate, SwapGate, XGate, ) GATES = [ (lambda: XXPlusYYGate(np.random.uniform(0, 2 * np.pi), np.random.uniform(0, 2 * np.pi)), 2), (lambda: PhaseGate(np.random.uniform(0, 2 * np.pi)), 1), (lambda: RZGate(np.random.uniform(0, 2 * np.pi)), 1), (lambda: CPhaseGate(np.random.uniform(0, 2 * np.pi)), 2), (lambda: SwapGate(), 2), (lambda: XGate(), 1), ] np.random.seed(0) n_qubits = 8 qc = QuantumCircuit(n_qubits) for _ in range(120): factory, nq = GATES[np.random.randint(len(GATES))] gate = factory() qubits = np.random.choice(n_qubits, size=nq, replace=False).tolist() qc.append(gate, qubits) from qiskit.quantum_info import SparsePauliOp observable = SparsePauliOp.from_list([("X" + "I" * (n_qubits - 1), 1.0)])

Propagating and writing to disk in one step¶

We'll use PauliPropagator and specify a file name to save to disk.

In [2]:

from propaq.circuits import PauliCircuit
from propaq.datatypes import PauliTermSum
from propaq.propagators import PauliPropagator

obs_term_sum = PauliTermSum.from_sparse_pauli_op(observable)
pc = PauliCircuit.from_qiskit(qc)

prop = PauliPropagator(n_threads=4, progress_bar=True)
propagated = prop.propagate(obs_term_sum, pc, filename="propagated_terms.gz")

print("Number of terms:", len(propagated.items()))
print("norm_squared:   ", propagated.norm_squared())

from propaq.circuits import PauliCircuit from propaq.datatypes import PauliTermSum from propaq.propagators import PauliPropagator obs_term_sum = PauliTermSum.from_sparse_pauli_op(observable) pc = PauliCircuit.from_qiskit(qc) prop = PauliPropagator(n_threads=4, progress_bar=True) propagated = prop.propagate(obs_term_sum, pc, filename="propagated_terms.gz") print("Number of terms:", len(propagated.items())) print("norm_squared: ", propagated.norm_squared())

Propagating:   0%|          | 0/281 [00:00<?, ?gate/s]

Number of terms: 32752
norm_squared:    0.9999999999999999

Reading it back eagerly¶

PauliTermSum.from_file loads the whole file into a fresh PauliTermSum in one call.

In [3]:

reloaded = PauliTermSum.from_file("propagated_terms.gz")

print("Number of terms:", len(reloaded.items()))
print("norm_squared:   ", reloaded.norm_squared())
print("Exactly matches the original:", dict(reloaded.items()) == dict(propagated.items()))

reloaded = PauliTermSum.from_file("propagated_terms.gz") print("Number of terms:", len(reloaded.items())) print("norm_squared: ", reloaded.norm_squared()) print("Exactly matches the original:", dict(reloaded.items()) == dict(propagated.items()))

Number of terms: 32752
norm_squared:    1.000000000000007
Exactly matches the original: True

Reading it back lazily¶

PauliTermStreamer.from_file opens the same file but yields one (term, coefficient) pair at a time as you iterate.

In [4]:

from propaq.datatypes import PauliTermStreamer

streamer = PauliTermStreamer.from_file("propagated_terms.gz")
print("First 5 terms in the file:")
for i, (term, coeff) in enumerate(streamer):
    if i >= 5:
        break
    print(f"  weight={term.weight:<2d} coeff={coeff}")

from propaq.datatypes import PauliTermStreamer streamer = PauliTermStreamer.from_file("propagated_terms.gz") print("First 5 terms in the file:") for i, (term, coeff) in enumerate(streamer): if i >= 5: break print(f" weight={term.weight:<2d} coeff={coeff}")

First 5 terms in the file:
  weight=8  coeff=2.3699622512925507e-05
  weight=6  coeff=0.0007565433568017365
  weight=7  coeff=-0.0007067802853712357
  weight=4  coeff=-0.0002886329864650659
  weight=5  coeff=7.070547645060666e-05

You can also merge a streamed file directly into an existing PauliTermSum with merge_from_file, which will lazily merge the two term sums.

In [5]:

streamed_accumulator = PauliTermSum()
streamed_accumulator.merge_from_file(PauliTermStreamer.from_file("propagated_terms.gz"))

print("Number of terms:", len(streamed_accumulator.items()))
print("norm_squared:   ", streamed_accumulator.norm_squared())
print("Exactly matches the original:", dict(streamed_accumulator.items()) == dict(propagated.items()))

streamed_accumulator = PauliTermSum() streamed_accumulator.merge_from_file(PauliTermStreamer.from_file("propagated_terms.gz")) print("Number of terms:", len(streamed_accumulator.items())) print("norm_squared: ", streamed_accumulator.norm_squared()) print("Exactly matches the original:", dict(streamed_accumulator.items()) == dict(propagated.items()))

Number of terms: 32752
norm_squared:    1.000000000000007
Exactly matches the original: True