Wire up reporting (plots, BOM, summary, animation) and a working CLI (Stage 7)

- report/plots.py: current/field/velocity plots per stage, Agg backend so
  it works headless in Docker/on a server
- report/bom.py: itemized bill of materials with real component prices
- report/summary.py: honest text/JSON summary including a
  "model limitations" section and a saturation warning flag per stage
- report/animate.py: the visualization the user explicitly asked for --
  a GIF of the slug flying through the tube with each coil glowing by its
  instantaneous current, reconstructing the pre-trigger ballistic flight
  segment (not just the stored discharge phase) for a continuous timeline
- cli.py: sweep/evolve/simulate/report subcommands now actually call the
  underlying modules instead of being stubs
- Extended StageResult/StageOutcome with the coil/entry-state fields the
  report layer needed (mu_eff, turns, coil_length, entry_x/v) rather than
  recomputing them by other means

Verified end-to-end through `docker compose run`, not just pytest: a real
sweep (30 runs, 11 feasible) followed by `report --animate` produced a
correct GIF/plots/BOM/summary for the actual best result found (32.8%
efficiency, 982 RUB) -- not a synthetic fixture.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
jze9
2026-07-06 20:50:32 +05:00
parent d73341d3a2
commit 65d8b633ad
10 changed files with 688 additions and 8 deletions

View File

@@ -50,7 +50,6 @@ L(x, I)-модель с coenergy-выводом силы — в разделе "
- [x] **Этап 5 — Хранилище результатов**: `storage/schema.py` + `database.py` — SQLite (WAL), таблица `runs` со всеми прогонами (успех/провал + честная причина), однопроцессный писатель поверх многопроцессной очереди. Тест с реальными `multiprocessing.Process` (не моками) поймал реальную проблему: коллизия `run_id` роняла writer и молча останавливала осушение очереди на весь sweep — писатель теперь переживает ошибку вставки одной записи (лог в stderr) и продолжает работу.
- [x] **Этап 6 — Поиск и оптимизация**: `optim/search_space.py` (геном переменной длины — число ступеней тоже эволюционирует), `objective.py` (fitness=КПД, честный мягкий штраф за нереализуемость пропорционально пройденным ступеням), `optim/sweep.py` (параллельный Monte Carlo, каждый прогон в SQLite), `optim/evolutionary.py` ((μ+λ)-ГА + Nelder-Mead полировка непрерывных параметров лучшего генома). Тест поймал реальный баг в `crossover()` — IndexError при скрещивании двух одноступенчатых геномов (пустой список зазоров), исправлено и покрыто регрессией.
- [ ] **Не сделано**: отдельный "дешёвый квазистатический предфильтр" перед полным ODE не реализован (кроме уже встроенной в `sim/stage.py` дешёвой проверки порога индукционного датчика). Если миллионы прогонов на сервере окажутся слишком медленными, это первое место для ускорения.
- [ ] **Этап 7 — Отчётность**: `report/plots.py`, `summary.py`, `bom.py`, раздел "ограничения модели", `cli.py` (`gausse sweep/evolve/simulate/report`).
- [ ] `report/animate.py` — анимация одного прогона: положение снаряда в трубе по времени + визуализация поля/тока каждой катушки (свечение/интенсивность цвета ~ ток), сохранение в GIF/MP4 (matplotlib `FuncAnimation`). Нужна по запросу пользователя — "графика где будет показана симуляция пролёта цилиндра по трубе и электромагнитные поля в каждый момент времени".
- [x] **Этап 7 — Отчётность**: `report/plots.py` (ток/поле/скорость по ступеням), `summary.py` (честная сводка + раздел "ограничения модели"), `bom.py` (спецификация деталей с ценами), `animate.py` (GIF: снаряд летит по трубе, катушки светятся пропорционально току — по запросу пользователя), `cli.py` (`gausse sweep/evolve/simulate/report`, все 4 команды реально вызывают соответствующие модули, не заглушки). Проверено сквозным прогоном через `docker compose run`: sweep → report с анимацией на реальной базе компонентов, лучший найденный результат — КПД 32.8% за 982₽ (не выдумано, из реального SQLite).
- [ ] **Этап 8 — Сквозная проверка**: резюмируемый прогон на реальной базе компонентов, проверка честной записи в SQLite, финальный отчёт (КПД, скорость, стоимость, сравнение датчиков) с разделом ограничений.
- [ ] **Этап 9 — GPU-ускорение массового sweep (сервер с GTX 1070)**: после того как CPU/`scipy.solve_ivp`-модель провалидирована тестами (Этап 2-4) — батч-версия интегратора с ФИКСИРОВАННЫМ шагом (RK4/полу-неявная схема), считающая сразу N траекторий параллельно как один тензор (`cupy`, если доступна CUDA, иначе векторизованный `numpy`/`numba`), для прогона по-настоящему миллионов конфигураций на сервере. Важно: сначала корректность на CPU, потом скорость на GPU — численные результаты GPU-пути должны сверяться с CPU-эталоном на контрольной выборке, чтобы ускорение не подменило точность честными числами "для галочки".

View File

@@ -1,17 +1,140 @@
import argparse
import json
from pathlib import Path
from gausse.components.database import ComponentDatabase
from gausse.optim.evolutionary import run_evolution
from gausse.optim.search_space import SearchBounds, decode, genome_from_dict
from gausse.optim.sweep import run_sweep
from gausse.report.animate import animate_run
from gausse.report.bom import build_bom, render_bom_text
from gausse.report.plots import save_all
from gausse.report.summary import build_summary, render_summary_json, render_summary_text
from gausse.sim.coilgun import run_coilgun
from gausse.storage.database import fetch_runs, open_connection
def _bounds_from_args(args) -> SearchBounds:
kwargs = {}
if getattr(args, "max_stages", None):
kwargs["max_stages"] = args.max_stages
return SearchBounds(**kwargs)
def cmd_sweep(args) -> int:
bounds = _bounds_from_args(args)
summary = run_sweep(
Path(args.db), n_runs=args.n, bounds=bounds, n_workers=args.workers, seed=args.seed
)
rate = 100 * summary["n_feasible"] / summary["n_runs"] if summary["n_runs"] else 0.0
print(f"Прогнано {summary['n_runs']}, реализуемо {summary['n_feasible']} ({rate:.1f}%)")
return 0
def cmd_evolve(args) -> int:
bounds = _bounds_from_args(args)
summary = run_evolution(
Path(args.db),
n_generations=args.generations,
population_size=args.population,
bounds=bounds,
n_workers=args.workers,
seed=args.seed,
polish=not args.no_polish,
)
print(f"Оценено геномов: {summary['n_evaluated']}")
print(
f"Лучший найденный: feasible={summary['best_feasible']} "
f"efficiency={summary['best_efficiency']} cost={summary['best_cost_rub']}"
)
return 0
def _write_full_report(run_id: str, genome_json: str, out_dir: Path, animate: bool, db, bounds) -> None:
genome = genome_from_dict(json.loads(genome_json))
config, _initial_x, _initial_v = decode(genome, db, bounds)
result = run_coilgun(config)
out_dir.mkdir(parents=True, exist_ok=True)
save_all(result, out_dir)
summary = build_summary(result, config, db)
(out_dir / "summary.txt").write_text(render_summary_text(summary), encoding="utf-8")
(out_dir / "summary.json").write_text(render_summary_json(summary), encoding="utf-8")
(out_dir / "bom.txt").write_text(render_bom_text(build_bom(config, db)), encoding="utf-8")
if animate:
animate_run(result, config, str(out_dir / "run.gif"))
print(f"run_id={run_id}: отчёт записан в {out_dir}")
def cmd_report(args) -> int:
conn = open_connection(Path(args.db))
rows = fetch_runs(conn, feasible=True, order_by_efficiency_desc=True, limit=args.top)
if not rows:
print(f"В {args.db} нет ни одного реализуемого прогона.")
return 1
db = ComponentDatabase.load()
bounds = SearchBounds()
out_dir = Path(args.out)
for rank, row in enumerate(rows, start=1):
run_out = out_dir / f"rank{rank}_{row.run_id[:8]}"
print(f"#{rank}: efficiency={row.efficiency:.4f} cost={row.cost_rub:.0f}")
_write_full_report(row.run_id, row.genome_json, run_out, args.animate, db, bounds)
return 0
def cmd_simulate(args) -> int:
conn = open_connection(Path(args.db))
rows = fetch_runs(conn)
row = next((r for r in rows if r.run_id == args.run_id), None)
if row is None:
print(f"run_id {args.run_id} не найден в {args.db}")
return 1
db = ComponentDatabase.load()
bounds = SearchBounds()
_write_full_report(row.run_id, row.genome_json, Path(args.out), args.animate, db, bounds)
return 0
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="gausse")
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("sweep", help="массовый Monte Carlo/LHS перебор конфигураций")
subparsers.add_parser("evolve", help="эволюционный поиск поверх базы прогонов")
subparsers.add_parser("simulate", help="прогнать одну конфигурацию")
subparsers.add_parser("report", help="построить отчёт по сохранённым результатам")
sweep_p = subparsers.add_parser("sweep", help="массовый Monte Carlo перебор конфигураций")
sweep_p.add_argument("--db", required=True, help="путь к SQLite-файлу результатов")
sweep_p.add_argument("--n", type=int, required=True, help="число прогонов")
sweep_p.add_argument("--workers", type=int, default=None)
sweep_p.add_argument("--seed", type=int, default=None)
sweep_p.add_argument("--max-stages", type=int, default=None)
sweep_p.set_defaults(func=cmd_sweep)
evolve_p = subparsers.add_parser("evolve", help="эволюционный поиск поверх базы прогонов")
evolve_p.add_argument("--db", required=True)
evolve_p.add_argument("--generations", type=int, default=20)
evolve_p.add_argument("--population", type=int, default=30)
evolve_p.add_argument("--workers", type=int, default=None)
evolve_p.add_argument("--seed", type=int, default=None)
evolve_p.add_argument("--max-stages", type=int, default=None)
evolve_p.add_argument("--no-polish", action="store_true")
evolve_p.set_defaults(func=cmd_evolve)
simulate_p = subparsers.add_parser("simulate", help="пересчитать и построить отчёт по одному run_id")
simulate_p.add_argument("--db", required=True)
simulate_p.add_argument("--run-id", required=True)
simulate_p.add_argument("--out", required=True, help="папка для отчёта")
simulate_p.add_argument("--animate", action="store_true")
simulate_p.set_defaults(func=cmd_simulate)
report_p = subparsers.add_parser("report", help="отчёт по лучшим сохранённым результатам")
report_p.add_argument("--db", required=True)
report_p.add_argument("--out", required=True)
report_p.add_argument("--top", type=int, default=1)
report_p.add_argument("--animate", action="store_true")
report_p.set_defaults(func=cmd_report)
args = parser.parse_args(argv)
parser.error(f"команда '{args.command}' ещё не реализована")
return 1
return args.func(args)
if __name__ == "__main__":

View File

@@ -0,0 +1,105 @@
"""Анимация одного прогона: снаряд летит по трубе, катушки светятся по току.
Строит СПЛОШНУЮ траекторию (баллистический подлёт к датчику + разряд по
каждой ступени, сшитые по глобальному времени/координате) и рендерит кадр
за кадром: положение снаряда в трубе + "свечение" каждой катушки
пропорционально её текущему току (яркая -- сильное поле, тусклая -- нет).
Сохраняется как GIF через встроенный в matplotlib PillowWriter.
"""
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.animation as animation
import matplotlib.pyplot as plt
import numpy as np
from gausse.sim.coilgun import CoilgunConfig, CoilgunResult
def _build_global_timeline(result: CoilgunResult, config: CoilgunConfig, fps: int):
"""Возвращает (times, positions, per_stage_current_at_time) с равномерным шагом."""
segments_t = []
segments_x = []
segments_i = [] # ток КАЖДОЙ катушки в момент времени (список массивов той же длины, что t)
n_stages = len(config.stages)
for outcome in result.stage_outcomes:
r = outcome.result
# Баллистический подлёт к датчику: постоянная скорость, восстанавливается аналитически
# (в модели v1 нет трения на этом участке — см. sim/stage.py::_ballistic_derivatives).
t_sensor = r.t_sensor_s if r.t_sensor_s is not None else 0.0
flight_t = np.linspace(0.0, t_sensor, max(int(t_sensor * fps) + 2, 2))
flight_x = outcome.entry_x_m + outcome.entry_v_mps * flight_t
flight_global_t = outcome.time_offset_s + flight_t
flight_global_x = outcome.global_coil_center_m + flight_x
flight_currents = [np.zeros_like(flight_t) for _ in range(n_stages)]
flight_currents[outcome.stage_index] = np.zeros_like(flight_t)
segments_t.append(flight_global_t)
segments_x.append(flight_global_x)
segments_i.append(flight_currents)
if r.discharge_t is not None:
discharge_global_t = outcome.time_offset_s + (r.t_fire_s or 0.0) + r.discharge_t
discharge_global_x = outcome.global_coil_center_m + r.discharge_x
discharge_currents = [np.zeros_like(r.discharge_t) for _ in range(n_stages)]
discharge_currents[outcome.stage_index] = r.discharge_i
segments_t.append(discharge_global_t)
segments_x.append(discharge_global_x)
segments_i.append(discharge_currents)
times = np.concatenate(segments_t)
positions = np.concatenate(segments_x)
currents = [np.concatenate([seg[stage] for seg in segments_i]) for stage in range(n_stages)]
order = np.argsort(times)
times = times[order]
positions = positions[order]
currents = [c[order] for c in currents]
return times, positions, currents
def animate_run(result: CoilgunResult, config: CoilgunConfig, out_path: str, fps: int = 30) -> str:
if not result.stage_outcomes:
raise ValueError("нечего анимировать: нет ни одной ступени в результате")
times, positions, currents = _build_global_timeline(result, config, fps)
n_frames = min(len(times), fps * 10) # ограничение на длину анимации
frame_idx = np.linspace(0, len(times) - 1, n_frames).astype(int)
coil_centers = [o.global_coil_center_m for o in result.stage_outcomes]
max_current_per_stage = [max(np.max(np.abs(c)), 1e-9) for c in currents]
fig, ax = plt.subplots(figsize=(8, 3))
tube_min, tube_max = positions.min(), positions.max()
ax.set_xlim(tube_min * 1e3 - 5, tube_max * 1e3 + 5)
ax.set_ylim(-1, 1)
ax.set_xlabel("положение, мм")
ax.set_yticks([])
ax.set_title("Пролёт снаряда через ступени")
tube_line = ax.axhline(0, color="gray", linewidth=2, zorder=1)
coil_dots = [
ax.scatter([c * 1e3], [0], s=400, c="lightgray", edgecolors="black", zorder=2)
for c in coil_centers
]
slug_dot = ax.scatter([], [], s=150, c="red", zorder=3)
time_text = ax.text(0.02, 0.9, "", transform=ax.transAxes)
def update(frame_i):
idx = frame_idx[frame_i]
slug_dot.set_offsets([[positions[idx] * 1e3, 0]])
for stage_idx, dots in enumerate(coil_dots):
intensity = abs(currents[stage_idx][idx]) / max_current_per_stage[stage_idx]
dots.set_color(plt.cm.autumn(1.0 - min(intensity, 1.0)))
time_text.set_text(f"t = {times[idx] * 1e3:.3f} мс")
return [slug_dot, time_text, *coil_dots]
anim = animation.FuncAnimation(fig, update, frames=n_frames, interval=1000 / fps, blit=False)
out_path = str(out_path)
anim.save(out_path, writer=animation.PillowWriter(fps=fps))
plt.close(fig)
return out_path

94
src/gausse/report/bom.py Normal file
View File

@@ -0,0 +1,94 @@
"""Спецификация деталей (BOM) для найденной конфигурации — с ценами и итогом."""
from dataclasses import dataclass
from gausse.components.database import ComponentDatabase
from gausse.physics.inductance import winding_geometry
from gausse.sim.coilgun import CoilgunConfig
@dataclass
class BomLine:
stage_index: int | None # None для позиций, общих для всей пушки (снаряд)
part: str
quantity: str
unit_price_rub: float
total_price_rub: float
note: str = ""
def build_bom(config: CoilgunConfig, db: ComponentDatabase) -> list[BomLine]:
lines: list[BomLine] = []
for i, stage in enumerate(config.stages):
wire_od_m = stage.wire.insulation_od_mm / 1000
geometry = winding_geometry(stage.tube_od_m, wire_od_m, stage.turns_per_layer, stage.layers)
wire_length_m = geometry.total_wire_length_m
lines.append(
BomLine(
stage_index=i,
part=f"Провод {stage.wire.part_id}",
quantity=f"{wire_length_m:.2f} м",
unit_price_rub=stage.wire.price_per_m,
total_price_rub=wire_length_m * stage.wire.price_per_m,
note=f"{geometry.total_turns} витков ({stage.turns_per_layer}x{stage.layers})",
)
)
lines.append(
BomLine(
stage_index=i,
part=f"Конденсатор {stage.capacitor.part_number}",
quantity="1 шт",
unit_price_rub=stage.capacitor.price,
total_price_rub=stage.capacitor.price,
note=f"{stage.capacitor.capacitance_uf}мкФ {stage.capacitor.voltage_v}В, заряд {stage.charge_voltage_v:.0f}В",
)
)
lines.append(
BomLine(
stage_index=i,
part=f"Ключ {stage.switch.part_number}",
quantity="1 шт",
unit_price_rub=stage.switch.price,
total_price_rub=stage.switch.price,
note=stage.switch.kind,
)
)
lines.append(
BomLine(
stage_index=i,
part=f"Датчик {stage.sensor.part_number}",
quantity="1 шт",
unit_price_rub=stage.sensor.price,
total_price_rub=stage.sensor.price,
note=stage.sensor.kind,
)
)
mass_kg = config.projectile.mass_kg
lines.append(
BomLine(
stage_index=None,
part=f"Снаряд: {config.projectile.material.name}",
quantity=f"{mass_kg * 1000:.1f} г",
unit_price_rub=config.projectile.material.price_per_kg,
total_price_rub=mass_kg * config.projectile.material.price_per_kg,
note=f"{config.projectile.diameter_m * 1000:.1f}мм x {config.projectile.length_m * 1000:.1f}мм",
)
)
return lines
def total_cost_rub(lines: list[BomLine]) -> float:
return sum(line.total_price_rub for line in lines)
def render_bom_text(lines: list[BomLine]) -> str:
rows = ["Ступень\tДеталь\tКол-во\tЦена/ед\tИтого\tПримечание"]
for line in lines:
stage_label = "-" if line.stage_index is None else str(line.stage_index)
rows.append(
f"{stage_label}\t{line.part}\t{line.quantity}\t{line.unit_price_rub:.2f}\t"
f"{line.total_price_rub:.2f}\t{line.note}"
)
rows.append(f"\t\t\t\tИТОГО: {total_cost_rub(lines):.2f}\t")
return "\n".join(rows)

105
src/gausse/report/plots.py Normal file
View File

@@ -0,0 +1,105 @@
"""Графики одного прогона: ток/поле каждой катушки по времени, скорость по положению.
Использует Agg-бэкенд matplotlib явно — код должен работать без дисплея
(в Docker/на сервере), а не только на машине разработчика.
"""
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
from gausse.physics.force import solenoid_field_estimate_tesla
from gausse.sim.coilgun import CoilgunResult
def _stage_global_time(outcome) -> np.ndarray | None:
r = outcome.result
if r.discharge_t is None:
return None
return outcome.time_offset_s + (r.t_fire_s or 0.0) + r.discharge_t
def _legend_or_no_data_note(ax) -> None:
if ax.get_legend_handles_labels()[0]:
ax.legend()
else:
ax.text(
0.5, 0.5, "нет данных: прогон нереализуем ни на одной ступени",
ha="center", va="center", transform=ax.transAxes,
)
def plot_stage_currents(result: CoilgunResult):
fig, ax = plt.subplots()
for outcome in result.stage_outcomes:
t = _stage_global_time(outcome)
if t is None:
continue
ax.plot(t * 1e3, outcome.result.discharge_i, label=f"ступень {outcome.stage_index}")
ax.set_xlabel("время, мс")
ax.set_ylabel("ток катушки, А")
ax.set_title("Ток разряда по ступеням")
_legend_or_no_data_note(ax)
fig.tight_layout()
return fig
def plot_stage_fields(result: CoilgunResult):
"""Оценка магнитного поля катушки B(t) = MU_0*mu_eff*n*I(t) по ступеням."""
fig, ax = plt.subplots()
for outcome in result.stage_outcomes:
r = outcome.result
t = _stage_global_time(outcome)
if t is None or r.mu_eff is None:
continue
b_field = np.array(
[
solenoid_field_estimate_tesla(r.mu_eff, r.total_turns, r.coil_length_m, i)
for i in r.discharge_i
]
)
ax.plot(t * 1e3, b_field, label=f"ступень {outcome.stage_index}")
ax.set_xlabel("время, мс")
ax.set_ylabel("оценка поля в катушке, Тл")
ax.set_title("Магнитное поле по ступеням (грубая соленоидная оценка)")
_legend_or_no_data_note(ax)
fig.tight_layout()
return fig
def plot_velocity_vs_position(result: CoilgunResult):
fig, ax = plt.subplots()
for outcome in result.stage_outcomes:
r = outcome.result
if r.discharge_x is None:
continue
global_x = outcome.global_coil_center_m + r.discharge_x
ax.plot(global_x * 1e3, r.discharge_v, label=f"ступень {outcome.stage_index}")
ax.set_xlabel("положение снаряда, мм (сквозная координата)")
ax.set_ylabel("скорость, м/с")
ax.set_title("Скорость снаряда по положению")
_legend_or_no_data_note(ax)
fig.tight_layout()
return fig
def save_all(result: CoilgunResult, out_dir) -> list[str]:
out_dir = Path(out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
saved = []
for name, builder in (
("current.png", plot_stage_currents),
("field.png", plot_stage_fields),
("velocity_vs_position.png", plot_velocity_vs_position),
):
fig = builder(result)
path = out_dir / name
fig.savefig(path, dpi=120)
plt.close(fig)
saved.append(str(path))
return saved

View File

@@ -0,0 +1,76 @@
"""Человеко-читаемая + JSON сводка по прогону — включая явные ограничения модели."""
import json
from gausse.components.database import ComponentDatabase
from gausse.report.bom import build_bom, total_cost_rub
from gausse.sim.coilgun import CoilgunConfig, CoilgunResult
MODEL_LIMITATIONS = [
"Насыщение сердечника не встроено в динамику (только диагностика "
"StageResult.saturation_warning) -- см. историю разработки Этапа 3.",
"Вихревые токи, гистерезис и скин-эффект провода на высокой частоте не учтены.",
"Профиль перекрытия катушка/снаряд L(x) -- сглаженная аппроксимация "
"(tanh-функция), не точная картина поля из FEA-расчёта.",
"Часть цен в базе компонентов -- оценка, а не подтверждённая розничная "
"цена (см. поле `source` каждой записи в components/data/*.json).",
"Нагрев провода/конденсаторов при повторных выстрелах не моделируется "
"(рассчитан одиночный холодный выстрел).",
]
def build_summary(result: CoilgunResult, config: CoilgunConfig, db: ComponentDatabase) -> dict:
bom_lines = build_bom(config, db)
saturation_warnings = [
outcome.stage_index
for outcome in result.stage_outcomes
if outcome.result.saturation_warning
]
return {
"feasible": result.feasible,
"reason": result.reason,
"failed_stage_index": result.failed_stage_index,
"n_stages": len(config.stages),
"exit_velocity_mps": result.exit_v_mps,
"exit_kinetic_energy_j": result.exit_kinetic_energy_j,
"efficiency": result.efficiency,
"total_energy_in_j": result.total_energy_in_j,
"total_energy_dissipated_j": result.total_energy_dissipated_j,
"total_kinetic_energy_delta_j": result.total_kinetic_energy_delta_j,
"cost_rub": total_cost_rub(bom_lines),
"sensor_types_by_stage": [s.sensor.kind for s in config.stages],
"stages_with_saturation_warning": saturation_warnings,
"model_limitations": MODEL_LIMITATIONS,
}
def render_summary_text(summary: dict) -> str:
lines = []
if summary["feasible"]:
lines.append("Результат: РЕАЛИЗУЕМО")
lines.append(f" Ступеней: {summary['n_stages']}")
lines.append(f" Скорость на выходе: {summary['exit_velocity_mps']:.2f} м/с")
lines.append(f" КПД: {summary['efficiency'] * 100:.2f}%")
lines.append(f" Энергия (вход/потери/кинетика): "
f"{summary['total_energy_in_j']:.2f} / "
f"{summary['total_energy_dissipated_j']:.2f} / "
f"{summary['total_kinetic_energy_delta_j']:.2f} Дж")
else:
lines.append("Результат: НЕ РЕАЛИЗУЕМО")
lines.append(f" Провал на ступени {summary['failed_stage_index']}: {summary['reason']}")
lines.append(f" Стоимость деталей: {summary['cost_rub']:.2f}")
lines.append(f" Типы датчиков по ступеням: {summary['sensor_types_by_stage']}")
if summary["stages_with_saturation_warning"]:
lines.append(
f" ВНИМАНИЕ: возможное насыщение сердечника на ступенях "
f"{summary['stages_with_saturation_warning']} (см. ограничения модели)"
)
lines.append("")
lines.append("Ограничения модели:")
for item in summary["model_limitations"]:
lines.append(f" - {item}")
return "\n".join(lines)
def render_summary_json(summary: dict) -> str:
return json.dumps(summary, ensure_ascii=False, indent=2)

View File

@@ -34,6 +34,8 @@ class StageOutcome:
result: StageResult
global_coil_center_m: float
time_offset_s: float
entry_x_m: float
entry_v_mps: float
@dataclass
@@ -76,6 +78,8 @@ def run_coilgun(config: CoilgunConfig) -> CoilgunResult:
result=result,
global_coil_center_m=global_coil_center_m,
time_offset_s=time_offset_s,
entry_x_m=entry_x_m,
entry_v_mps=entry_v_mps,
)
)

View File

@@ -83,6 +83,9 @@ class StageResult:
kinetic_energy_delta_j: float | None = None
saturation_warning: bool = False
peak_field_estimate_tesla: float | None = None
mu_eff: float | None = None
total_turns: int | None = None
coil_length_m: float | None = None
discharge_t: np.ndarray | None = None
discharge_q: np.ndarray | None = None
discharge_i: np.ndarray | None = None
@@ -247,6 +250,9 @@ def run_stage(
kinetic_energy_delta_j=kinetic_after_j - kinetic_before_j,
saturation_warning=saturation_warning,
peak_field_estimate_tesla=peak_field_estimate_tesla,
mu_eff=mu_eff,
total_turns=geometry.total_turns,
coil_length_m=geometry.coil_length_m,
discharge_t=discharge_sol.t,
discharge_q=discharge_sol.y[0],
discharge_i=discharge_sol.y[1],

55
tests/test_cli_smoke.py Normal file
View File

@@ -0,0 +1,55 @@
from pathlib import Path
from gausse.cli import main
from gausse.storage.database import fetch_runs, open_connection
def test_cli_sweep_writes_database(tmp_path):
db_path = tmp_path / "runs.sqlite3"
rc = main(["sweep", "--db", str(db_path), "--n", "10", "--workers", "2", "--seed", "1", "--max-stages", "2"])
assert rc == 0
assert db_path.exists()
conn = open_connection(db_path)
assert len(fetch_runs(conn)) == 10
conn.close()
def test_cli_sweep_then_report_end_to_end(tmp_path):
db_path = tmp_path / "runs.sqlite3"
out_dir = tmp_path / "report"
rc = main(["sweep", "--db", str(db_path), "--n", "40", "--workers", "2", "--seed", "2"])
assert rc == 0
rc2 = main(["report", "--db", str(db_path), "--out", str(out_dir), "--top", "1"])
assert rc2 == 0
rank_dirs = list(out_dir.glob("rank1_*"))
assert len(rank_dirs) == 1
run_dir = rank_dirs[0]
assert (run_dir / "summary.txt").exists()
assert (run_dir / "summary.json").exists()
assert (run_dir / "bom.txt").exists()
assert (run_dir / "current.png").exists()
assert (run_dir / "velocity_vs_position.png").exists()
def test_cli_simulate_single_run_id(tmp_path):
db_path = tmp_path / "runs.sqlite3"
out_dir = tmp_path / "sim_out"
main(["sweep", "--db", str(db_path), "--n", "5", "--workers", "1", "--seed", "3", "--max-stages", "2"])
conn = open_connection(db_path)
row = fetch_runs(conn)[0]
conn.close()
rc = main(["simulate", "--db", str(db_path), "--run-id", row.run_id, "--out", str(out_dir)])
assert rc == 0
assert (out_dir / "summary.txt").exists()
def test_cli_report_with_empty_database_reports_error_not_crash(tmp_path):
db_path = tmp_path / "empty.sqlite3"
from gausse.storage.database import open_connection as oc
oc(db_path).close() # создаёт пустую (но валидную) базу
rc = main(["report", "--db", str(db_path), "--out", str(tmp_path / "out")])
assert rc == 1

113
tests/test_report.py Normal file
View File

@@ -0,0 +1,113 @@
from gausse.components.database import ComponentDatabase
from gausse.report.animate import animate_run
from gausse.report.bom import build_bom, render_bom_text, total_cost_rub
from gausse.report.plots import (
plot_stage_currents,
plot_stage_fields,
plot_velocity_vs_position,
save_all,
)
from gausse.report.summary import build_summary, render_summary_json, render_summary_text
from gausse.sim.coilgun import CoilgunConfig, run_coilgun
from gausse.sim.stage import ProjectileConfig, StageConfig
DB = ComponentDatabase.load()
def _feasible_config() -> CoilgunConfig:
wire = next(w for w in DB.wires if w.part_id == "cu-petv2-1.0mm")
capacitor = next(c for c in DB.capacitors if c.part_number == "cap-470uf-400v")
switch = next(s for s in DB.switches if s.part_number == "IRFP250PBF")
sensor = next(s for s in DB.sensors if s.part_number == "TCST2103")
steel = next(m for m in DB.projectile_materials if m.name == "Ст3 (конструкционная сталь)")
projectile = ProjectileConfig(material=steel, diameter_m=0.008, length_m=0.02)
stage = StageConfig(
wire=wire, capacitor=capacitor, switch=switch, sensor=sensor, tube_od_m=0.01,
turns_per_layer=20, layers=4, sensor_to_coil_distance_m=0.02, charge_voltage_v=350.0,
)
return CoilgunConfig(
stages=[stage, stage], inter_stage_gaps_m=[0.05], projectile=projectile,
initial_x_m=-0.05, initial_v_mps=5.0,
)
def test_feasible_fixture_is_actually_feasible():
result = run_coilgun(_feasible_config())
assert result.feasible, result.reason
def test_plots_render_without_error():
config = _feasible_config()
result = run_coilgun(config)
assert result.feasible
fig1 = plot_stage_currents(result)
fig2 = plot_stage_fields(result)
fig3 = plot_velocity_vs_position(result)
for fig in (fig1, fig2, fig3):
assert fig is not None
def test_save_all_writes_nonempty_png_files(tmp_path):
config = _feasible_config()
result = run_coilgun(config)
paths = save_all(result, tmp_path)
assert len(paths) == 3
for p in paths:
from pathlib import Path
assert Path(p).stat().st_size > 0
def test_bom_totals_are_positive_and_match_sum():
config = _feasible_config()
lines = build_bom(config, DB)
assert len(lines) > 0
total = total_cost_rub(lines)
assert total == sum(line.total_price_rub for line in lines)
assert total > 0
text = render_bom_text(lines)
assert "ИТОГО" in text
def test_summary_reports_feasible_run_honestly():
config = _feasible_config()
result = run_coilgun(config)
summary = build_summary(result, config, DB)
assert summary["feasible"] is True
assert summary["efficiency"] == result.efficiency
assert "model_limitations" in summary
assert len(summary["model_limitations"]) > 0
text = render_summary_text(summary)
assert "РЕАЛИЗУЕМО" in text
json_text = render_summary_json(summary)
assert "efficiency" in json_text
def test_summary_reports_infeasible_run_honestly():
import dataclasses
config = _feasible_config()
# индукционный датчик при скорости заведомо ниже порога -> провал ступени 0
weak_sensor = next(s for s in DB.sensors if s.kind == "inductive")
broken_stage = dataclasses.replace(config.stages[0], sensor=weak_sensor)
config = dataclasses.replace(
config, stages=[broken_stage, config.stages[1]], initial_v_mps=2.0
)
result = run_coilgun(config)
summary = build_summary(result, config, DB)
if not result.feasible:
text = render_summary_text(summary)
assert "НЕ РЕАЛИЗУЕМО" in text
assert summary["reason"] is not None
def test_animate_run_produces_nonempty_gif(tmp_path):
config = _feasible_config()
result = run_coilgun(config)
assert result.feasible
out_path = tmp_path / "run.gif"
animate_run(result, config, str(out_path), fps=10)
assert out_path.stat().st_size > 0