Initial commit: планирование эксперимента и оптимизация смесей

This commit is contained in:
2026-07-23 15:12:32 +05:00
commit 025421193d
17 changed files with 1634 additions and 0 deletions
View File
+183
View File
@@ -0,0 +1,183 @@
# core/classes.py
"""
Модуль содержит классы для представления ингредиентов и смеси.
"""
from dataclasses import dataclass
class Ingredient:
"""Представляет ингредиент с именем, единицей измерения и значением."""
def __init__(self, name: str = "Water", unit: str = "ml", value: float = 1.0):
self.name = name
self.unit = unit
self.value = value
def to_dict(self):
return {"name": self.name, "unit": self.unit, "value": self.value}
@staticmethod
def from_dict(data):
return Ingredient(data["name"], data["unit"], data["value"])
def __repr__(self):
return f"{self.name}: {self.value} {self.unit}"
@dataclass
class IngredientInfo:
"""
Обёртка для ингредиента в составе смеси, добавляет разбавление и плотность.
"""
ingredient: Ingredient
dilution: float = 1.0
density: float = 1.0
def to_dict(self):
return {
"ingredient": self.ingredient.to_dict(),
"dilution": self.dilution,
"density": self.density
}
@staticmethod
def from_dict(data):
ing = Ingredient.from_dict(data["ingredient"])
return IngredientInfo(
ingredient=ing,
dilution=data.get("dilution", 1.0),
density=data.get("density", 1.0)
)
@property
def name(self):
return self.ingredient.name
@property
def unit(self):
return self.ingredient.unit
@property
def value(self):
return self.ingredient.value
@value.setter
def value(self, new_value):
self.ingredient.value = new_value
class Mixture:
"""
Основной класс для работы со смесью.
Первый ингредиент считается растворителем (его процент пересчитывается автоматически).
"""
def __init__(self, solvent=None, ings=None, total_amount=1000.0, total_unit="ml"):
all_ings = []
if solvent is None:
water = Ingredient(name="Water", unit=total_unit, value=100.0)
solvent_info = IngredientInfo(ingredient=water)
elif isinstance(solvent, Ingredient):
solvent_info = IngredientInfo(ingredient=solvent)
elif isinstance(solvent, IngredientInfo):
solvent_info = solvent
else:
raise TypeError("solvent must be Ingredient or IngredientInfo")
all_ings.append(solvent_info)
if ings is not None:
if not isinstance(ings, list):
raise TypeError("ings must be a list")
for item in ings:
if isinstance(item, Ingredient):
all_ings.append(IngredientInfo(ingredient=item))
elif isinstance(item, IngredientInfo):
all_ings.append(item)
else:
raise TypeError("Each item in ings must be Ingredient or IngredientInfo")
self.ings = all_ings
self.total_amount = total_amount
self.total_unit = total_unit
self.comp_solvent_percent()
def comp_solvent_percent(self) -> float:
"""Пересчёт процента растворителя = 100% - сумма остальных."""
if not self.ings:
return 0.0
total_others = sum(info.value for info in self.ings[1:])
if total_others > 100.0:
raise ValueError("Sum of all ingredient percentages exceeds 100%")
self.ings[0].value = 100.0 - total_others
return self.ings[0].value
def add_ingredient(self, ingredient: Ingredient, dilution: float = 1.0, density: float = 1.0):
info = IngredientInfo(ingredient, dilution, density)
self.ings.append(info)
self.comp_solvent_percent()
def remove_ingredient(self, index: int):
if index == 0:
raise ValueError("Нельзя удалить растворитель (первый ингредиент).")
removed = self.ings.pop(index)
self.comp_solvent_percent()
return removed
def get_absolute_ingredients(self):
"""Возвращает абсолютные значения ингредиентов (с учётом плотности и разбавления)."""
abs_list = []
sum_rest = 0.0
for info in self.ings[1:]:
abs_value = (info.value / 100.0) * self.total_amount * info.dilution * info.density
abs_list.append(Ingredient(name=info.name, unit=info.unit, value=abs_value))
sum_rest += abs_value
if sum_rest > self.total_amount:
raise ValueError("Сумма ингредиентов (кроме растворителя) превышает общее количество")
solvent_abs = self.total_amount - sum_rest
solvent = Ingredient(name=self.ings[0].name, unit=self.ings[0].unit, value=solvent_abs)
abs_list.insert(0, solvent)
return abs_list
def get_ingredient_info(self, index: int):
return self.ings[index]
def set_total_amount(self, new_amount: float, new_unit: str):
if new_amount <= 0:
raise ValueError("Общее количество должно быть положительным.")
if not new_unit.strip():
raise ValueError("Единица измерения не может быть пустой.")
self.total_amount = new_amount
self.total_unit = new_unit.strip()
def set_composition(self, *percentages):
expected = len(self.ings) - 1
if len(percentages) != expected:
raise ValueError(f"Expected {expected} percentages, got {len(percentages)}")
total = sum(percentages)
if total > 100.0:
raise ValueError(f"Sum of percentages ({total:.2f}) exceeds 100")
for i, pct in enumerate(percentages, start=1):
if pct < 0:
raise ValueError(f"Percentage for {self.ings[i].name} cannot be negative")
self.ings[i].value = pct
self.comp_solvent_percent()
def to_dict(self):
return {
"ings": [info.to_dict() for info in self.ings],
"total_amount": self.total_amount,
"total_unit": self.total_unit
}
@staticmethod
def from_dict(data):
ings = [IngredientInfo.from_dict(item) for item in data["ings"]]
solvent = ings[0] if ings else None
rest = ings[1:] if len(ings) > 1 else None
return Mixture(
solvent=solvent,
ings=rest,
total_amount=data["total_amount"],
total_unit=data["total_unit"]
)
def __repr__(self):
parts = []
for info in self.ings:
parts.append(f"{info.name} {info.value:.1f}% (разб.{info.dilution:.2f}, плотн.{info.density:.2f})")
return f"Mixture({', '.join(parts)}, total={self.total_amount} {self.total_unit})"
+137
View File
@@ -0,0 +1,137 @@
# core/doe.py
"""
Модуль с функциями для генерации планов эксперимента.
"""
import random
import numpy as np
def ispoz(N):
"""Проверяет, является ли N положительным целым числом (для использования в ze)."""
return isinstance(N, (int, float)) and not isinstance(N, bool) and N > 0
def ze(k, n):
"""Генерирует матрицу размера n x k, заполненную нулями (центральные точки)."""
if n is None or k is None:
return []
elif ispoz(k) and ispoz(n):
return [[0] * k for _ in range(n)]
return []
def ffe(k):
"""Полный факторный план 2^k. Возвращает список списков размером 2^k x k со значениями ±1."""
if k < 2:
raise ValueError("Factor must be at least 2")
N = 2 ** k
out = []
for i in range(N):
row = []
for j in range(k):
bit = ((i >> j) & 1) * 2 - 1
row.append(bit)
out.append(row)
return out
def ccp(k, alpha=1.0, center_points=1, fractional=None):
"""
Центральный композиционный план (ЦКП) для k факторов.
alpha – звёздное плечо.
center_points – количество центральных точек.
fractional – степень дробления ядра (если None, используется полный факторный план).
"""
if k < 2:
raise ValueError("Для ЦКП необходимо минимум 2 фактора.")
if fractional is not None:
if fractional <= 0 or fractional >= k:
raise ValueError("Степень дробления p должна быть от 1 до k-1")
base_k = k - fractional
if base_k < 2:
raise ValueError(f"Ядро дробного факторного плана должно иметь минимум 2 фактора, а получается {base_k}.")
if fractional is not None:
from core.doe import fractional_factorial
core_plan = fractional_factorial(k, fractional)
else:
core_plan = ffe(k)
star_points = []
for i in range(k):
row = [0.0] * k
row[i] = alpha
star_points.append(row)
row_neg = [0.0] * k
row_neg[i] = -alpha
star_points.append(row_neg)
center_rows = [[0.0] * k for _ in range(center_points)]
plan = core_plan + star_points + center_rows
return plan
def fractional_factorial(k, p=1):
"""Дробный факторный план 2^(k-p). p – степень дробления."""
if k < 2 or p <= 0 or p >= k:
raise ValueError("Некорректные параметры: k>=2, 0<p<k")
base_k = k - p
if base_k < 2:
raise ValueError(f"Ядро должно содержать минимум 2 фактора, а получается {base_k}.")
N = 2 ** base_k
base_plan = ffe(base_k)
plan = []
for row in base_plan:
new_row = row.copy()
for i in range(p):
prod = 1
for j in range(base_k):
prod *= row[j]
new_row.append(prod)
plan.append(new_row)
return plan
def plackett_burman(k):
"""
План Плакетта-Бермана для k факторов.
Число опытов N – ближайшее кратное 4, >= k+1.
Для N = 8, 16, 32 используется рекурсивное построение матрицы Адамара.
"""
if k < 2:
raise ValueError("Требуется минимум 2 фактора.")
N = 4 * ((k + 3) // 4)
def hadamard_rec(n):
if n == 1:
return np.array([[1]])
H = hadamard_rec(n//2)
top = np.hstack([H, H])
bottom = np.hstack([H, -H])
return np.vstack([top, bottom])
if (N & (N - 1)) == 0: # N — степень двойки
H = hadamard_rec(N)
else:
try:
from scipy.linalg import hadamard
H = hadamard(N)
except ImportError:
print("Предупреждение: scipy не установлена. Генерируем случайный план.")
plan = []
for _ in range(N):
row = [random.choice([-1, 1]) for _ in range(k)]
plan.append(row)
return plan
plan = H[:, 1:k+1].tolist()
return plan
def box_behnken(k):
"""
План Бокса-Бенкена для k факторов (k >= 3).
Содержит комбинации на серединах рёбер и центральные точки.
"""
if k < 3:
raise ValueError("Требуется минимум 3 фактора.")
plan = []
for i in range(k):
for j in range(i+1, k):
for a, b in [(1,1), (1,-1), (-1,1), (-1,-1)]:
row = [0.0] * k
row[i] = a
row[j] = b
plan.append(row)
for _ in range(3):
plan.append([0.0] * k)
return plan
+99
View File
@@ -0,0 +1,99 @@
# core/experiment.py
"""
Класс ExperimentDesign для хранения шагов и матрицы плана,
а также генерации экспериментальных смесей.
"""
from core.classes import Mixture, Ingredient, IngredientInfo
class ExperimentDesign:
def __init__(self, steps=None, plan=None):
self.steps = steps if steps is not None else []
self.plan = plan if plan is not None else []
def set_steps(self, steps):
"""
Устанавливает шаги для всех факторов (ингредиентов кроме растворителя).
Каждый шаг должен быть >= 0. Нулевые шаги означают, что данный ингредиент не является фактором.
"""
for s in steps:
if s < 0:
raise ValueError("Step cannot be negative")
self.steps = list(steps)
def _active_indices(self):
"""Возвращает список индексов (в self.steps) с шагом > 0."""
return [i for i, s in enumerate(self.steps) if s > 0]
def _active_steps(self):
"""Возвращает список шагов > 0."""
return [s for s in self.steps if s > 0]
def generate_plan(self, k=None):
"""
Генерирует полный факторный план 2^k, где k - число активных факторов (шагов > 0).
Если k не указано, вычисляется автоматически.
Если активных факторов < 2, выбрасывает ValueError.
"""
active = self._active_steps()
if k is None:
k = len(active)
if k < 2:
raise ValueError(f"Need at least 2 active factors (steps > 0), but got {k}")
from core.doe import ffe
self.plan = ffe(k)
def get_experiment_mixture(self, mixture, exp_index):
"""
Возвращает Mixture для опыта с индексом exp_index (0-based).
Изменяются только те ингредиенты, у которых шаг > 0.
"""
if not self.plan or not self.steps:
raise ValueError("Plan or steps not set")
if exp_index >= len(self.plan):
raise IndexError("Experiment index out of range")
active_indices = self._active_indices()
active_steps = self._active_steps()
if len(active_indices) != len(self.plan[0]):
raise ValueError("Number of active steps does not match plan columns")
base_percents = [info.value for info in mixture.ings]
new_percents = base_percents.copy()
for col, idx_in_steps in enumerate(active_indices):
factor = self.plan[exp_index][col]
ingredient_idx = idx_in_steps + 1
delta = active_steps[col] * factor
new_percents[ingredient_idx] = base_percents[ingredient_idx] + delta
new_ings = []
for i, info in enumerate(mixture.ings):
new_ings.append(
IngredientInfo(
ingredient=Ingredient(name=info.name, unit=info.unit, value=new_percents[i]),
dilution=info.dilution,
density=info.density
)
)
temp_mixture = Mixture(
solvent=new_ings[0],
ings=new_ings[1:],
total_amount=mixture.total_amount,
total_unit=mixture.total_unit
)
return temp_mixture
def get_all_mixtures(self, mixture):
"""Возвращает список Mixture для всех опытов."""
return [self.get_experiment_mixture(mixture, i) for i in range(len(self.plan))]
def to_dict(self):
return {
"steps": self.steps,
"plan": self.plan
}
@classmethod
def from_dict(cls, data):
return cls(steps=data.get("steps", []), plan=data.get("plan", []))
+47
View File
@@ -0,0 +1,47 @@
# core/fs.py
"""
Модуль для работы с файлами проекта (сохранение/загрузка в JSON).
"""
import json
from core.classes import Mixture
from core.experiment import ExperimentDesign
PROJECT_FILENAME = "project.json"
def load_project(filename: str = PROJECT_FILENAME):
"""
Загружает проект из JSON-файла.
Возвращает кортеж (mixture, design, responses).
Если файл отсутствует или повреждён, возвращает новые объекты по умолчанию.
"""
try:
with open(filename, "r", encoding="utf-8") as f:
data = json.load(f)
mixture = Mixture.from_dict(data.get("mixture", {})) if "mixture" in data else Mixture()
design = ExperimentDesign.from_dict(data.get("design", {})) if "design" in data else ExperimentDesign()
responses = data.get("responses", [])
if isinstance(responses, list) and responses and isinstance(responses[0], (int, float)):
responses = [{"name": "Отклик 1", "values": responses}]
return mixture, design, responses
except (FileNotFoundError, json.JSONDecodeError, KeyError):
return Mixture(), ExperimentDesign(), []
def save_project_partial(filename: str = PROJECT_FILENAME, mixture=None, design=None, responses=None):
"""
Сохраняет только переданные части проекта, остальные загружаются из существующего файла.
Используется для точечного сохранения (например, только смеси).
"""
cur_mixture, cur_design, cur_responses = load_project(filename)
if mixture is not None:
cur_mixture = mixture
if design is not None:
cur_design = design
if responses is not None:
cur_responses = responses
data = {
"mixture": cur_mixture.to_dict(),
"design": cur_design.to_dict(),
"responses": cur_responses
}
with open(filename, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
+130
View File
@@ -0,0 +1,130 @@
# core/statistics.py
"""
Статистические функции для регрессии, проверки значимости, дисперсии и градиента.
"""
import numpy as np
try:
from scipy import stats
SCIPY_AVAILABLE = True
except ImportError:
SCIPY_AVAILABLE = False
def compute_regression(X, y):
"""МНК-регрессия, возвращает коэффициенты, ошибки, t-статистики."""
N, p = X.shape
try:
XTX = X.T @ X
XTX_inv = np.linalg.inv(XTX)
beta = XTX_inv @ (X.T @ y)
y_pred = X @ beta
residuals = y - y_pred
df_res = N - p
if df_res > 0:
s2_res = np.sum(residuals**2) / df_res
se = np.sqrt(np.diag(XTX_inv) * s2_res)
else:
s2_res = np.nan
se = [np.nan] * p
t_stats = beta / se if not np.isnan(se).any() else [np.nan] * p
except np.linalg.LinAlgError:
beta = [np.nan] * p
se = [np.nan] * p
t_stats = [np.nan] * p
df_res = 0
s2_res = np.nan
residuals = np.full(N, np.nan)
y_pred = np.full(N, np.nan)
XTX_inv = None
return {
"beta": beta,
"se": se,
"t_stats": t_stats,
"df_res": df_res,
"s2_res": s2_res,
"residuals": residuals,
"y_pred": y_pred,
"XTX_inv": XTX_inv
}
def regression_summary(y, y_pred, k, df_res):
"""Вычисляет R², скорректированный R², F-статистику и p-value."""
n = len(y)
ss_total = np.sum((y - np.mean(y))**2)
ss_res = np.sum((y - y_pred)**2)
r2 = 1 - ss_res / ss_total if ss_total > 0 else np.nan
if df_res > 0:
r2_adj = 1 - (1 - r2) * (n - 1) / df_res
else:
r2_adj = np.nan
if df_res > 0 and k > 0:
f_stat = (r2 / k) / ((1 - r2) / df_res) if r2 < 1 else np.inf
else:
f_stat = np.nan
if SCIPY_AVAILABLE and not np.isnan(f_stat) and f_stat != np.inf:
p_value = 1 - stats.f.cdf(f_stat, k, df_res)
else:
p_value = np.nan
return {"r2": r2, "r2_adj": r2_adj, "f_stat": f_stat, "f_pvalue": p_value}
def get_center_indices(plan, active_indices):
"""Возвращает индексы строк плана, где активные факторы равны 0."""
if not plan:
return []
plan_arr = np.array(plan)
if active_indices:
active_cols = plan_arr[:, active_indices]
center_mask = np.all(active_cols == 0, axis=1)
return np.where(center_mask)[0].tolist()
return []
def compute_center_stats(y_center):
"""Среднее и несмещённая дисперсия для центральных точек."""
if len(y_center) < 2:
return None, None
return np.mean(y_center), np.var(y_center, ddof=1)
def chi2_confidence_interval(var, n, alpha=0.05):
"""Доверительный интервал для дисперсии на основе χ²-распределения."""
if not SCIPY_AVAILABLE or n < 2:
return np.nan, np.nan
df = n - 1
chi2_lower = stats.chi2.ppf(alpha / 2, df)
chi2_upper = stats.chi2.ppf(1 - alpha / 2, df)
return df * var / chi2_upper, df * var / chi2_lower
def get_t_critical(df, alpha=0.05):
"""Критическое значение t (двустороннее)."""
if not SCIPY_AVAILABLE or df <= 0:
return np.nan
return stats.t.ppf(1 - alpha/2, df)
def compute_gradient_points(beta, base_levels, step_factor, num_steps, factor_names=None, maximize=True):
"""
Генерирует точки вдоль градиента (крутое восхождение или спуск).
beta – коэффициенты модели: [b0, b1, b2, ...]
base_levels – список базовых уровней для каждого фактора
step_factor – шаг по первому фактору
num_steps – количество шагов
factor_names – список названий факторов
maximize – True для максимизации, False для минимизации
"""
if len(beta) - 1 != len(base_levels):
raise ValueError("Число коэффициентов (без b0) должно совпадать с числом факторов")
direction = 1 if maximize else -1
coefs = np.array(beta[1:]) * direction
max_abs = np.max(np.abs(coefs))
if max_abs == 0:
raise ValueError("Все коэффициенты равны нулю, направление не определено.")
norm_coefs = coefs / max_abs
steps = step_factor * norm_coefs
points = []
for step_num in range(1, num_steps + 1):
point = {}
for i, name in enumerate(factor_names or [f"x{i+1}" for i in range(len(base_levels))]):
point[name] = base_levels[i] + steps[i] * step_num
predicted = beta[0] + np.sum(beta[1:] * np.array([point[name] for name in factor_names or range(len(base_levels))]))
point["Шаг"] = step_num
point["Предсказанный отклик"] = predicted
points.append(point)
return points
+87
View File
@@ -0,0 +1,87 @@
# core/table_io.py
"""
Модуль для работы с таблицами: вывод, сохранение в CSV, копирование в буфер обмена.
"""
import os
import csv
from table import print_table
try:
import pyperclip
HAS_PYPERCLIP = True
except ImportError:
HAS_PYPERCLIP = False
def handle_table_output(headers, rows, title="Таблица", ask_each_time=True):
"""
Выводит таблицу, затем предлагает сохранить в CSV или скопировать в буфер обмена.
headers – список заголовков
rows – список строк
title – заголовок
ask_each_time если False, только выводит таблицу без дополнительных вопросов.
"""
if title:
print(f"\n=== {title} ===")
print_table(headers, rows)
if not ask_each_time:
return
while True:
print("\nДействия с таблицей:")
print(" 1. Сохранить в CSV")
if HAS_PYPERCLIP:
print(" 2. Скопировать в буфер обмена")
else:
print(" 2. Скопировать в буфер обмена (недоступно, установите pyperclip)")
print(" Enter - пропустить")
choice = input("Выберите действие: ").strip()
if choice == "":
break
if choice == "1":
save_to_csv(headers, rows)
break
elif choice == "2":
if HAS_PYPERCLIP:
copy_to_clipboard(headers, rows)
break
else:
print("pyperclip не установлен. Установите командой: pip install pyperclip")
continue
else:
print("Неверный выбор. Введите 1, 2 или Enter.")
def save_to_csv(headers, rows, default_filename="table.csv"):
"""Сохраняет таблицу в CSV-файл."""
filename = input(f"Имя файла (по умолчанию {default_filename}): ").strip()
if not filename:
filename = default_filename
if not filename.lower().endswith('.csv'):
filename += '.csv'
try:
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.writer(f, delimiter=';')
writer.writerow(headers)
writer.writerows(rows)
print(f"Таблица сохранена в {os.path.abspath(filename)}")
except Exception as e:
print(f"Ошибка при сохранении: {e}")
def copy_to_clipboard(headers, rows):
"""Копирует таблицу в буфер обмена в формате TSV."""
if not HAS_PYPERCLIP:
print("pyperclip не установлен. Установите командой: pip install pyperclip")
return
lines = []
lines.append("\t".join(str(h) for h in headers))
for row in rows:
lines.append("\t".join(str(cell) for cell in row))
text = "\n".join(lines)
try:
pyperclip.copy(text)
print("Таблица скопирована в буфер обмена (TSV-формат).")
print("Вставьте её в Excel или другой редактор.")
except Exception as e:
print(f"Ошибка копирования в буфер обмена: {e}")