src/ga_qas.py

@@ -3,7 +3,6 @@
 # "Genetic optimization of ansatz expressibility for enhanced variational quantum algorithm performance"
 
 import random
-import sys
 from typing import Generator, Never
 
 import matplotlib.pyplot as plt
@@ -12,6 +11,14 @@ from qas_flow import Stream
 from quantum_circuit import (Gate, GateType, QuantumCircuit, circ_from_layers,
                              sample_random_generator, single_typ)
 
+DEPTH: int = 6
+QUBITS: int = 6
+GENERATIONS: int = 40
+GENERATION_SIZE: int = 60
+PARENT_AMOUNT: int = 10
+MUTATION_RATE: float = 0.1
+
+
 gate_set: list[GateType] = [
     GateType.H,
     GateType.RX,
@@ -70,59 +77,47 @@ def sample_hyperspace(
         yield point
 
 
-EXPRESSIBILITY_SAMPLES: int = 2000
+def plot_best_circuits(best_circuits: list[QuantumCircuit]) -> None:
+    fig, ax = plt.subplots()
+
+    ax.plot([-circ.expressibility for circ in best_circuits])
+    fig.savefig("best_circuits.png")
 
 
-def run_ga_qas(
+def main() -> None:
-    depth: int,
+    seed_rng: random.Random = random.Random(1020381)
-    qubits: int,
-    generations: int,
-    generation_size: int,
-    parent_amount: int,
-    mutation_rate: float,
-    seed: int,
-) -> list[tuple[int, int, int, int, int, float, float, float]]:
-
-    print(
-        f"running GA QAS {seed} with {qubits} qubits, {depth} depth, for {generations} generations of size {generation_size}, with {parent_amount} parents and {mutation_rate:.3f} mutation rate",
-        file=sys.stderr,
-    )
-
-    seed_rng = random.Random(seed)
     initial_population: list[QuantumCircuit] = (
-        Stream(
+        Stream(sample_random_generator(random.Random(101020), QUBITS, DEPTH, gate_set))
-            sample_random_generator(
+        .apply(lambda circ: print(circ))
-                random.Random(seed_rng.randint(1000, 1000000000)),
-                qubits,
-                depth,
-                gate_set,
-            )
-        )
         .apply(
             lambda circ: circ.expressibility_estimate(
-                EXPRESSIBILITY_SAMPLES, seed_rng.randint(1000, 1000000000)
+                2000, seed_rng.randint(1000, 1000000000)
             )
         )
-        .take(generation_size)
+        .apply(lambda circ: print(circ))
+        .take(GENERATION_SIZE)
         .collect()
     )
 
-    population: list[QuantumCircuit] = initial_population
+    population = initial_population
-    main_rng = random.Random(seed_rng.randint(1000, 1000000000))
+
+    main_rng = random.Random(2837175)
 
     best_circuits: list[QuantumCircuit] = []
-    return_data: list[tuple[int, int, int, int, int, float, float, float]] = []
+
-    for generation in range(generations):
+    for generation in range(GENERATIONS):
-        print(f"starting generation {generation} for seed {seed}", file=sys.stderr)
+        print(f"starting generation {generation}")
         population.sort(key=lambda qc: qc.expressibility, reverse=True)
-        parents: list[QuantumCircuit] = population[:parent_amount]
+        parents = population[:PARENT_AMOUNT]
-        offspring: list[QuantumCircuit] = []
+        for parent in parents:
-        for _ in range(generation_size):
+            print(parent)
+        offspring = []
+        for _ in range(GENERATION_SIZE):
             [p1, p2] = main_rng.sample(parents, 2)
-            crossover_layer = main_rng.randint(1, depth)
+            crossover_layer = main_rng.randint(1, DEPTH)
             child_layers = p1.gates[:crossover_layer] + p2.gates[crossover_layer:]
-            if main_rng.random() < mutation_rate:
+            if main_rng.random() < MUTATION_RATE:
-                layer_idx = main_rng.randrange(depth)
+                layer_idx = main_rng.randrange(DEPTH)
                 layer = child_layers[layer_idx]
                 gate_idx = main_rng.randrange(len(layer))
                 old_gate = child_layers[layer_idx][gate_idx]
@@ -142,83 +137,21 @@ def run_ga_qas(
                         old_gate.param_idx,
                     )
 
-            child = circ_from_layers(child_layers, qubits)
+            child = circ_from_layers(child_layers, QUBITS)
             child.expressibility_estimate(2000, seed_rng.randint(1000, 1000000000))
             offspring.append(child)
 
         offspring.sort(key=lambda qc: qc.expressibility, reverse=True)
-        return_data.append(
-            (
-                depth,
-                qubits,
-                generation,
-                generation_size,
-                parent_amount,
-                mutation_rate,
-                population[0].expressibility,
-                offspring[0].expressibility,
-            )
-        )
         if population[0].expressibility > offspring[0].expressibility:
+            print(f"best parent > best child")
             best_circuits.append(population[0])
         else:
+            print(f"best child > best parent")
             best_circuits.append(offspring[0])
         population = offspring
-    print(f"finished seed {seed}", file=sys.stderr)
-    return return_data
 
-
+    plot_best_circuits(best_circuits)
-def run_from_point(pnt: tuple[tuple[int, int, int, int, float], int]):
+    plt.show()
-    (point, seed) = pnt
-    return run_ga_qas(
-        point[0],
-        point[1],
-        20,
-        point[2],
-        point[3],
-        point[4],
-        seed,
-    )
-
-
-def print_ret(ret_data):
-    for dat in ret_data:
-        (
-            depth,
-            qubits,
-            generation,
-            generation_size,
-            parent_amount,
-            mutation_rate,
-            best_pop,
-            best_offspring,
-        ) = dat
-        print(
-            f"{depth},{qubits},{generation},{generation_size},{parent_amount},{mutation_rate},{best_pop},{best_offspring}",
-            flush=True,
-        )
-
-
-def main() -> None:
-
-    rng = random.Random(123456789)
-
-    results: list[QuantumCircuit] = (
-        Stream(
-            sample_hyperspace(
-                (1, 40),
-                (1, 10),
-                (1, 100),
-                (1, 20),
-                (0.0, 1.0),
-                seed=rng.randint(1000, 1000000000),
-            )
-        )
-        .map(lambda point: (point, rng.randint(1000, 1000000000)))
-        .par_map(run_from_point)
-        .apply(print_ret)
-        .collect()
-    )
 
 
 if __name__ == "__main__":

src/qas_flow/funcs.py

@@ -1,9 +1,4 @@
-import os
+from typing import Callable
-import sys
-from collections.abc import Generator
-from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait
-from typing import Any, Callable, Never
-
 from .stream import Stream, T, U
 
 
@@ -13,48 +8,20 @@ def map(stream: Stream[T], fn: Callable[[T], U]) -> Stream[U]:
     Applies the function `fn` to every element of the stream
     """
 
-    def gen() -> Generator[U, Any, None]:
+    def gen():
         for x in stream:
             yield fn(x)
 
     return Stream(gen())
 
 
-@Stream.extension()
-def par_map(stream: Stream[T], fn: Callable[[T], U], cores: int = 0) -> Stream[U]:
-
-    def gen() -> Generator[U, Any, None]:
-        nprocs = (os.cpu_count() or 2) - 1 if cores <= 0 else cores
-        inflight = max(1, nprocs)
-
-        it = iter(stream)
-
-        with ProcessPoolExecutor(max_workers=nprocs) as ex:
-            pending = set()
-
-            for _ in range(inflight):
-                x = next(it)
-                pending.add(ex.submit(fn, x))
-
-            while True:
-                done, pending = wait(pending, return_when=FIRST_COMPLETED)
-
-                for fut in done:
-                    print(f"yielding: {fut}", file=sys.stderr)
-                    yield fut.result()
-                    x = next(it)
-                    pending.add(ex.submit(fn, x))
-
-    return Stream(gen())
-
-
 @Stream.extension()
 def filter(stream: Stream[T], pred: Callable[[T], bool]) -> Stream[T]:
     """
     Applies the predicate `pred` to every element and only returns elements where the predicate is true
     """
 
-    def gen() -> Generator[T, Any, None]:
+    def gen():
         for x in stream:
             if pred(x):
                 yield x
@@ -68,7 +35,7 @@ def apply(stream: Stream[T], fn: Callable[[T], None]) -> Stream[T]:
     Apply the function `fn` to every element of the stream in-place
     """
 
-    def gen() -> Generator[T, Any, None]:
+    def gen():
         for x in stream:
             fn(x)
             yield x
@@ -82,7 +49,7 @@ def take(stream: Stream[T], n: int) -> Stream[T]:
     Return a stream with at most `n` elements
     """
 
-    def gen() -> Generator[T, Any, None]:
+    def gen():
         c = 0
         for x in stream:
             if c < n:
@@ -100,7 +67,7 @@ def skip(stream: Stream[T], n: int) -> Stream[T]:
     Ignore the first `n` elements of a stream
     """
 
-    def gen() -> Generator[T, Any, None]:
+    def gen():
         c = 0
         for x in stream:
             c += 1
@@ -116,7 +83,7 @@ def batch(stream: Stream[T], n: int) -> Stream[list[T]]:
     Create batches of size `n` from the stream
     """
 
-    def gen() -> Generator[list[T], Any, None]:
+    def gen():
         ls: list[T] = []
         for x in stream:
             ls.append(x)
@@ -133,7 +100,7 @@ def enumerate(stream: Stream[T]) -> Stream[tuple[int, T]]:
     Add an index to each element of the stream
     """
 
-    def gen() -> Generator[tuple[int, T], Any, None]:
+    def gen():
         idx = 0
         for x in stream:
             yield (idx, x)

src/quantum_circuit.py

@@ -2,7 +2,7 @@ import math
 import random
 from dataclasses import dataclass
 from enum import IntEnum
-from typing import Self, override
+from typing import override
 
 import numpy as np
 from qiskit import QuantumCircuit as QiskitCircuit
@@ -167,7 +167,7 @@ class QuantumCircuit:
 
     def expressibility_estimate(
         self, samples: int, seed: int, bins: int = 75, eps: float = 1e-12
-    ) -> Self:
+    ) -> "QuantumCircuit":
         qc, thetas = self.to_qiskit_for_expressibility()
 
         if self.params <= 0: