GPU в эволюции: оценка поколения одним батчем (use_gpu=True)

evaluate_genomes_gpu в gpu/batch_sweep.py — общий раундовый симулятор
(_simulate_states) для sweep и эволюции, та же формула фитнеса, что в
objective.evaluate (КПД либо -1+доля пройденных ступеней). run_evolution
получил use_gpu: поколение целиком уходит в батч-интегратор (на сервере
cupy/GTX 1070), полировка остаётся точной на CPU. search_mode=evolve-gpu(...).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/gausse/gpu/batch_sweep.py

@@ -49,6 +49,7 @@ class _State:
     kinetic_delta_j: float = 0.0
     alive: bool = True
     reason: str | None = None
+    failed_stage_index: int | None = None
 
 
 def _coast_velocity(v0: float, distance_m: float, retard_const_n: float, drag_coeff: float, mass_kg: float) -> float:
@@ -101,6 +102,108 @@ def _prepare_stage(stage, projectile, cur_x_global, cur_v, coil_center_global):
     }, None
 
 
+def _simulate_states(xp, states: list) -> None:
+    """Раунд-за-раундом прогоняет разряды всех живых ступеней батчами (мутирует states)."""
+    if not states:
+        return
+    max_ns = max(len(s.config.stages) for s in states)
+    for s_idx in range(max_ns):
+        prepared = []  # (state, prep)
+        for st in states:
+            if not st.alive or len(st.config.stages) <= s_idx:
+                continue
+            p, reason = _prepare_stage(
+                st.config.stages[s_idx], st.config.projectile, st.cur_x, st.cur_v, st.coil_centers[s_idx]
+            )
+            if p is None:
+                st.alive = False
+                st.reason = f"ступень {s_idx}: {reason}"
+                st.failed_stage_index = s_idx
+            else:
+                prepared.append((st, p))
+        if not prepared:
+            continue
+
+        models = [p["phys"].inductance_model for _, p in prepared]
+        cps = [p["phys"].circuit_params for _, p in prepared]
+        params = params_from_models(xp, models, cps)
+        out = integrate_batch_discharge(
+            xp,
+            xp.asarray([p["q0"] for _, p in prepared]),
+            xp.asarray([p["x_fire"] for _, p in prepared]),
+            xp.asarray([p["v_fire"] for _, p in prepared]),
+            params, dt=2e-6, max_steps=15000,
+        )
+        exit_v = to_cpu(xp, out["exit_v"]); exit_x = to_cpu(xp, out["exit_x"])
+        peak_i = to_cpu(xp, out["peak_current"]); feas = to_cpu(xp, out["feasible"])
+        for k, (st, p) in enumerate(prepared):
+            stage = p["stage"]
+            if not bool(feas[k]):
+                st.alive = False
+                st.reason = f"ступень {s_idx}: разряд не скоммутировался"
+                st.failed_stage_index = s_idx
+                continue
+            surge = switch_pulse_limit_a(stage.switch)
+            if float(peak_i[k]) > surge:
+                st.alive = False
+                st.reason = f"ступень {s_idx}: пиковый ток {float(peak_i[k]):.0f}А > импульсного предела ключа ({surge:.0f}А)"
+                st.failed_stage_index = s_idx
+                continue
+            ev = float(exit_v[k])
+            st.kinetic_delta_j += 0.5 * p["mass"] * (ev**2 - p["v_fire"] ** 2)
+            st.energy_in_j += p["energy_in"]
+            st.cur_v = ev
+            st.cur_x = p["coil_center"] + float(exit_x[k])
+
+
+def evaluate_genomes_gpu(xp, genomes: list, db: ComponentDatabase, bounds: SearchBounds) -> list:
+    """Оценка списка геномов ОДНИМ батчем (для эволюции): та же физика и та же
+    формула фитнеса, что в objective.evaluate (КПД либо -1+доля пройденных
+    ступеней), но разряды всех геномов интегрируются вместе на GPU/numpy.
+    """
+    from gausse.optim.objective import EvaluationResult, compute_cost_rub
+
+    states = []
+    for g in genomes:
+        cfg, ix, iv = decode(g, db, bounds)
+        centers = [0.0]
+        for gap in cfg.inter_stage_gaps_m:
+            centers.append(centers[-1] + gap)
+        states.append(_State(g, cfg, ix, iv, centers, cur_x=ix, cur_v=iv))
+
+    _simulate_states(xp, states)
+
+    results = []
+    for st in states:
+        cost = compute_cost_rub(st.config, db)
+        ok = st.alive and st.energy_in_j > 0
+        detail = build_detail(
+            st.config, _FEASIBLE if ok else None, db, st.initial_x_m, st.initial_v_mps,
+            st.genome.tube_inner_d_m, st.genome.tube_wall_m,
+        )
+        if ok:
+            eff = st.kinetic_delta_j / st.energy_in_j
+            results.append(EvaluationResult(
+                feasible=True, cost_rub=cost, fitness=eff, efficiency=eff,
+                exit_velocity_mps=st.cur_v,
+                energy_breakdown={
+                    "total_energy_in_j": st.energy_in_j,
+                    "total_kinetic_energy_delta_j": st.kinetic_delta_j,
+                },
+                detail=detail,
+            ))
+        else:
+            n_stages = len(st.config.stages)
+            progress = (st.failed_stage_index or 0) / n_stages if n_stages else 0.0
+            results.append(EvaluationResult(
+                feasible=False, cost_rub=cost, fitness=-1.0 + progress,
+                reason=st.reason or "нет ни одной сработавшей ступени",
+                failed_stage_index=st.failed_stage_index,
+                energy_breakdown={}, detail=detail,
+            ))
+    return results
+
+
 def run_gpu_sweep(
     db_path: Path,
     n_runs: int,
@@ -131,49 +234,7 @@ def run_gpu_sweep(
                     centers.append(centers[-1] + gap)
                 states.append(_State(g, cfg, ix, iv, centers, cur_x=ix, cur_v=iv))
 
-            max_ns = max(len(s.config.stages) for s in states)
-            for s_idx in range(max_ns):
-                prepared = []  # (state, prep)
-                for st in states:
-                    if not st.alive or len(st.config.stages) <= s_idx:
-                        continue
-                    p, reason = _prepare_stage(
-                        st.config.stages[s_idx], st.config.projectile, st.cur_x, st.cur_v, st.coil_centers[s_idx]
-                    )
-                    if p is None:
-                        st.alive = False
-                        st.reason = f"ступень {s_idx}: {reason}"
-                    else:
-                        prepared.append((st, p))
-                if not prepared:
-                    continue
-
-                models = [p["phys"].inductance_model for _, p in prepared]
-                cps = [p["phys"].circuit_params for _, p in prepared]
-                params = params_from_models(xp, models, cps)
-                out = integrate_batch_discharge(
-                    xp,
-                    xp.asarray([p["q0"] for _, p in prepared]),
-                    xp.asarray([p["x_fire"] for _, p in prepared]),
-                    xp.asarray([p["v_fire"] for _, p in prepared]),
-                    params, dt=2e-6, max_steps=15000,
-                )
-                exit_v = to_cpu(xp, out["exit_v"]); exit_x = to_cpu(xp, out["exit_x"])
-                peak_i = to_cpu(xp, out["peak_current"]); feas = to_cpu(xp, out["feasible"])
-                for k, (st, p) in enumerate(prepared):
-                    stage = p["stage"]
-                    if not bool(feas[k]):
-                        st.alive = False; st.reason = f"ступень {s_idx}: разряд не скоммутировался"; continue
-                    surge = switch_pulse_limit_a(stage.switch)
-                    if float(peak_i[k]) > surge:
-                        st.alive = False
-                        st.reason = f"ступень {s_idx}: пиковый ток {float(peak_i[k]):.0f}А > импульсного предела ключа ({surge:.0f}А)"
-                        continue
-                    ev = float(exit_v[k])
-                    st.kinetic_delta_j += 0.5 * p["mass"] * (ev**2 - p["v_fire"] ** 2)
-                    st.energy_in_j += p["energy_in"]
-                    st.cur_v = ev
-                    st.cur_x = p["coil_center"] + float(exit_x[k])
+            _simulate_states(xp, states)
 
             records = []
             for st in states:

src/gausse/optim/evolutionary.py

@@ -7,6 +7,7 @@
 компонентов.
 """
 
+import contextlib
 import copy
 import multiprocessing
 import random
@@ -96,13 +97,25 @@ def run_evolution(
     mutation_rate: float = 0.2,
     polish: bool = True,
     log_path: Path | None = None,
+    use_gpu: bool = False,
 ) -> dict:
     n_workers = n_workers or multiprocessing.cpu_count()
     rng = random.Random(seed)
     db = ComponentDatabase.load(data_dir) if data_dir else ComponentDatabase.load()
 
+    # GPU-режим: всё поколение оценивается ОДНИМ батчем (gpu/batch_sweep) —
+    # та же физика через общий build_stage_physics, фикс.шаг RK4 вместо
+    # адаптивного solve_ivp (сверено в тестах, ~3%). Полировка остаётся на CPU.
+    gpu_xp = None
+    backend = "cpu-pool"
+    if use_gpu:
+        from gausse.gpu.backend import get_backend
+
+        gpu_xp, backend = get_backend(prefer_gpu=True)
+
+    mode = "evolve" if gpu_xp is None else f"evolve-gpu({backend})"
     logger = ProgressLogger(
-        log_path or default_log_path(db_path), mode="evolve", total=n_generations * population_size
+        log_path or default_log_path(db_path), mode=mode, total=n_generations * population_size
     )
 
     ctx = multiprocessing.get_context("spawn")
@@ -115,18 +128,28 @@ def run_evolution(
     n_evaluated = 0
 
     try:
-        with ProcessPoolExecutor(
+        executor_cm = (
+            ProcessPoolExecutor(
                 max_workers=n_workers,
                 mp_context=ctx,
                 initializer=worker_context.init_worker,
                 initargs=(data_dir, bounds),
-        ) as executor:
+            )
+            if gpu_xp is None
+            else contextlib.nullcontext()
+        )
+        with executor_cm as executor:
             for generation in range(n_generations):
+                if gpu_xp is not None:
+                    from gausse.gpu.batch_sweep import evaluate_genomes_gpu
+
+                    pairs = list(zip(population, evaluate_genomes_gpu(gpu_xp, population, db, bounds)))
+                else:
                     pairs = list(executor.map(_evaluate_genome, population))
                 n_evaluated += len(pairs)
 
                 for genome, result in pairs:
-                    record: RunRecord = build_run_record(genome, result, db, search_mode="evolve")
+                    record: RunRecord = build_run_record(genome, result, db, search_mode=mode)
                     queue.put(record)
                     logger.update(result.feasible, result.efficiency)
                     if best_pair is None or result.fitness > best_pair[1].fitness:

tests/test_evolutionary.py

@@ -63,3 +63,19 @@ def test_polish_does_not_make_the_best_genome_worse():
 
     _, polished_result = polish_best(genome, DB, bounds)
     assert polished_result.fitness >= baseline_fitness - 1e-9
+
+
+def test_run_evolution_gpu_batch_mode(tmp_path):
+    """GPU-режим эволюции (батч-оценка поколения; здесь numpy-бэкенд):
+    пишет прогоны в ту же БД и находит реализуемые конфигурации."""
+    from gausse.storage.database import count_runs, fetch_runs, open_connection
+
+    db_path = tmp_path / "evo_gpu.sqlite3"
+    summary = run_evolution(
+        db_path, n_generations=3, population_size=40, seed=5, polish=False, use_gpu=True
+    )
+    assert summary["n_evaluated"] == 120
+    conn = open_connection(db_path)
+    assert count_runs(conn) >= 120
+    rows = fetch_runs(conn, limit=5)
+    assert all(r.search_mode.startswith("evolve-gpu") for r in rows)