""" core/formula_engine.py Детерминированный движок инженерных расчётов. Ключевые принципы: 1. Числовые расчёты выполняются детерминированно, без LLM. 2. QueryParser / extract_variables / extract_city — основной источник параметров. 3. Если все обязательные параметры уже есть, расчёт выполняется напрямую. 4. Таблицы используются только для добора климатических данных или справочных значений. 5. Движок совместим с более старой архитектурой проекта: - callbacks on_city_not_found / on_material_not_found - cache materials / cities - helper-методы build_*_response - reasoning chain """ from __future__ import annotations import logging import math import json import re from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Callable logger = logging.getLogger(__name__) try: from core.query_parser import extract_city, extract_variables, get_parameter_issues except ImportError: def get_parameter_issues(text: str, length_kind: str | None = None) -> list[str]: return ["Недоступна проверка единиц исходных данных; восстановите core.query_parser."] def extract_city(text: str) -> str | None: text = (text or "").strip() m = re.search(r"\bдля\s+([А-ЯA-ZЁ][А-ЯA-ZЁа-яa-zё\- ]{1,50})", text) return m.group(1).strip() if m else None def extract_variables(text: str) -> dict[str, float]: aliases = { "l": "L", "tv": "t_v", "t_v": "t_v", "tв": "t_v", "tn": "t_n", "t_n": "t_n", "tн": "t_n", "tot": "t_ot", "t_ot": "t_ot", "tот": "t_ot", "zot": "z_ot", "z_ot": "z_ot", "zот": "z_ot", "a": "A", "r": "R", "delta_t": "delta_t", "dt": "delta_t", "delta": "delta", "λ": "lambda_value", "lambda": "lambda_value", "lambda_value": "lambda_value", "r_tr": "R_tr", "rtr": "R_tr", "q": "Q", } normalized = ( (text or "") .lower() .replace(",", ".") .replace("°с", "") .replace("°c", "") .replace("м³", "м3") ) values: dict[str, float] = {} for alias, canonical in aliases.items(): pattern = rf"(?:\b{re.escape(alias)}\b)\s*=\s*(-?\d+(?:\.\d+)?)" m = re.search(pattern, normalized, flags=re.IGNORECASE) if m: values[canonical] = float(m.group(1)) return values try: from core.table_calculator import TableCalculator except ImportError: TableCalculator = None # type: ignore @dataclass(slots=True) class Material: name: str lambda_value: float density: float = 0.0 specific_heat: float | None = None source: str = "" @dataclass(slots=True) class CityClimate: name: str t_ot: float z_ot: int t_n: float source: str = "" @dataclass(slots=True) class FormulaMeta: id: str name: str expression: str description: str source: str unit: str required_params: list[str] climate_params: list[str] = field(default_factory=list) aliases: list[str] = field(default_factory=list) handler_name: str = "" class FormulaEngine: """ Выполняет инженерные расчёты и добирает климатические данные при необходимости. """ CLIMATE_DEPENDENT = frozenset({ "gsop", "ventilation_heat", "heat_loss", }) PARAM_LABELS = { "L": "L — расход воздуха, м³/ч", "t_v": "t_v — температура внутреннего воздуха, °C", "t_n": "t_n — температура наружного воздуха, °C", "t_ot": "t_ot — средняя температура отопительного периода, °C", "z_ot": "z_ot — продолжительность отопительного периода, сут", "A": "A — площадь ограждения, м²", "delta_t": "delta_t — разность температур, °C", "R": "R — сопротивление теплопередаче, м²·°C/Вт", "delta": "delta — толщина слоя, м", "lambda_value": "lambda_value — теплопроводность λ, Вт/(м·°C)", "R_tr": "R_tr — требуемое сопротивление, м²·°C/Вт", "Q": "Q — тепловая мощность, Вт", } CITY_ALIASES_COMMON = { "мск": "Москва", "москва": "Москва", "спб": "Санкт-Петербург", "питер": "Санкт-Петербург", "санкт петербург": "Санкт-Петербург", "екб": "Екатеринбург", "екатеринбург": "Екатеринбург", "нн": "Нижний Новгород", } MATERIAL_ALIASES_COMMON = { "минвата": "минеральная вата", "минеральная вата": "минеральная вата", "каменная вата": "минеральная вата", "базальтовая вата": "минеральная вата", "пеноплекс": "экструдированный пенополистирол", "эппс": "экструдированный пенополистирол", "xps": "экструдированный пенополистирол", "пенополистирол": "пенополистирол", "кирпич": "кирпич", "бетон": "бетон", "газобетон": "газобетон", } def __init__(self, qa_system: Any = None): self.qa_system = qa_system self.reasoning_steps: list[str] = [] self.materials: dict[str, Material] = {} self.cities: dict[str, CityClimate] = {} self._material_cache: dict[str, Material] = {} self._city_cache: dict[str, CityClimate] = {} self._table_calculator: Any = None self.on_city_not_found: Callable[[str], Any] | None = None self.on_material_not_found: Callable[[str], Any] | None = None self.formulas: dict[str, dict[str, Any]] = self._init_formulas() self.city_list: list[str] = [] self.city_aliases: dict[str, str] = {} self.material_aliases: dict[str, str] = dict(self.MATERIAL_ALIASES_COMMON) self._load_cache() self.city_list = sorted(self.cities.keys(), key=len, reverse=True) self.city_aliases = self._build_city_aliases() def _init_formulas(self) -> dict[str, dict[str, Any]]: return { "gsop": { "id": "gsop", "name": "ГСОП", "expression": "ГСОП = (t_в - t_от) × z_от", "description": "Градусо-сутки отопительного периода.", "legend": { "t_v": "температура внутреннего воздуха, °C", "t_ot": "средняя температура отопительного периода, °C", "z_ot": "продолжительность отопительного периода, сут", }, "source": "СП 131.13330", "unit": "°C·сут", "required_params": ["t_v", "t_ot", "z_ot"], "climate_params": ["t_ot", "z_ot"], "aliases": [ "гсоп", "градусо-сутки", "градусосутки", "градусо суток", "dd", ], "handler": self._calc_gsop, }, "ventilation_heat": { "id": "ventilation_heat", "name": "Расход теплоты на вентиляцию", "expression": "Q_в = 0.335 × L × (t_в - t_н)", "description": "Расчёт расхода теплоты на нагрев наружного воздуха.", "legend": { "L": "расход воздуха, м³/ч", "t_v": "температура внутреннего воздуха, °C", "t_n": "температура наружного воздуха, °C", }, "source": "СП 60.13330", "unit": "Вт", "required_params": ["L", "t_v", "t_n"], "climate_params": ["t_n"], "aliases": [ "вентиляция", "вентиляционный", "расход теплоты на вентиляцию", "теплота на вентиляцию", "теплопотери на вентиляцию", "приточный воздух", "приточная вентиляция", "нагрев воздуха", ], "handler": self._calc_ventilation_heat, }, "heat_loss": { "id": "heat_loss", "name": "Теплопотери через ограждение", "expression": "Q = (A × Δt) / R", "description": "Расчёт теплопотерь через ограждающую конструкцию.", "legend": { "A": "площадь ограждения, м²", "delta_t": "разность температур, °C", "R": "сопротивление теплопередаче, м²·°C/Вт", }, "source": "СП 50.13330", "unit": "Вт", "required_params": ["A", "delta_t", "R"], "climate_params": ["delta_t"], "aliases": [ "теплопотери", "потери тепла", "потери теплоты", "тепловые потери", "ограждение", "наружная стена", "потери через стену", ], "handler": self._calc_heat_loss, }, "thermal_resistance_layer": { "id": "thermal_resistance_layer", "name": "Сопротивление теплопередаче слоя", "expression": "R = δ / λ", "description": "Термическое сопротивление однородного слоя материала.", "legend": { "delta": "толщина слоя, м", "lambda_value": "теплопроводность λ, Вт/(м·°C)", }, "source": "СП 50.13330", "unit": "м²·°C/Вт", "required_params": ["delta", "lambda_value"], "climate_params": [], "aliases": [ "сопротивление слоя", "термическое сопротивление", "термическое сопротивление слоя", "r слоя", ], "handler": self._calc_thermal_resistance, }, "required_insulation_thickness": { "id": "required_insulation_thickness", "name": "Требуемая толщина изоляции", "expression": "δ = R_тр × λ", "description": "Оценка требуемой толщины утеплителя по требуемому сопротивлению.", "legend": { "R_tr": "требуемое сопротивление, м²·°C/Вт", "lambda_value": "теплопроводность λ, Вт/(м·°C)", }, "source": "СП 50.13330", "unit": "м", "required_params": ["R_tr", "lambda_value"], "climate_params": [], "aliases": [ "толщина утеплителя", "толщина изоляции", "требуемая толщина", "толщина теплоизоляции", ], "handler": self._calc_insulation_thickness, }, "pipe_surface_heat_flux": { "id": "pipe_surface_heat_flux", "name": "Удельный тепловой поток", "expression": "q = Q / L", "description": "Удельный тепловой поток на единицу длины.", "legend": { "Q": "тепловая мощность, Вт", "L": "длина, м", }, "source": "СП 61.13330", "unit": "Вт/м", "required_params": ["Q", "L"], "climate_params": [], "aliases": [ "удельный поток", "удельный тепловой поток", "тепловой поток трубы", "теплопоток", ], "handler": self._calc_pipe_heat_flux, }, } @staticmethod def _get_cache_path() -> Path: cache_dir = Path("cache") cache_dir.mkdir(parents=True, exist_ok=True) return cache_dir / "formula_engine_cache.json" def _load_cache(self) -> None: cache_path = self._get_cache_path() if not cache_path.exists(): return try: data = json.loads(cache_path.read_text(encoding="utf-8")) for key, mat_data in data.get("materials", {}).items(): material = Material( name=str(mat_data["name"]), lambda_value=float(mat_data["lambda_value"]), density=float(mat_data.get("density", 0.0)), specific_heat=mat_data.get("specific_heat"), source=str(mat_data.get("source", "")), ) self.materials[key] = material self._material_cache[key] = material for key, city_data in data.get("cities", {}).items(): city = CityClimate( name=str(city_data["name"]), t_ot=float(city_data["t_ot"]), z_ot=int(city_data["z_ot"]), t_n=float(city_data.get("t_n", city_data["t_ot"])), source=str(city_data.get("source", "")), ) self.cities[key] = city self._city_cache[key] = city except ( OSError, json.JSONDecodeError, KeyError, TypeError, ValueError, ) as exc: logger.warning("Не удалось загрузить кэш FormulaEngine: %s", exc) self.materials = {} self.cities = {} self._material_cache = {} self._city_cache = {} def _save_cache(self) -> None: try: payload = { "materials": { key: asdict(mat) for key, mat in self.materials.items() }, "cities": { key: asdict(city) for key, city in self.cities.items() }, } self._get_cache_path().write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) except OSError: logger.exception("Не удалось сохранить кэш FormulaEngine") def _build_city_aliases(self) -> dict[str, str]: aliases = dict(self.CITY_ALIASES_COMMON) for city_name, city in self.cities.items(): canonical = city.name or city_name lowered = city_name.lower().strip() aliases[lowered] = canonical aliases[canonical.lower().strip()] = canonical normalized = lowered.replace("-", " ").replace("ё", "е") aliases[normalized] = canonical return aliases @staticmethod def _normalize_text(text: str) -> str: text = text or "" text = text.lower() text = text.replace("ё", "е") text = text.replace(",", ".") text = text.replace("°с", "") text = text.replace("°c", "") text = text.replace("м³", "м3") text = text.replace("m³", "m3") text = text.replace("–", "-").replace("—", "-") text = re.sub(r"\s+", " ", text) return text.strip() @staticmethod def _merge_parameters( extracted: dict[str, float], provided: dict[str, Any], ) -> dict[str, float]: result: dict[str, float] = {} for key, value in extracted.items(): if isinstance(value, (int, float)) and not isinstance(value, bool): result[key] = float(value) for key, value in provided.items(): if isinstance(value, (int, float)) and not isinstance(value, bool): result[key] = float(value) # Проверка согласованности явного Δt выполняется перед расчётом. # Вывод из температур делаем после объединения, учитывая overrides API. if "delta_t" not in result and {"t_v", "t_n"} <= result.keys(): result["delta_t"] = result["t_v"] - result["t_n"] return result def _extract_parameters_from_text(self, text: str) -> dict[str, float]: values = dict(extract_variables(text) or {}) normalized = self._normalize_text(text) extra_patterns: list[tuple[str, str]] = [ (r"\bрасход воздуха\s*=?\s*(-?\d+(?:\.\d+)?)", "L"), (r"\bтемпература внутри(?:\s*воздуха)?\s*=?\s*(-?\d+(?:\.\d+)?)", "t_v"), (r"\bтемпература внутреннего воздуха\s*=?\s*(-?\d+(?:\.\d+)?)", "t_v"), (r"\bтемпература наружного воздуха\s*=?\s*(-?\d+(?:\.\d+)?)", "t_n"), (r"\bнаружная температура\s*=?\s*(-?\d+(?:\.\d+)?)", "t_n"), (r"\bплощадь\s*=?\s*(-?\d+(?:\.\d+)?)", "A"), (r"\bсопротивление\s*=?\s*(-?\d+(?:\.\d+)?)", "R"), (r"\bтолщина\s*=?\s*(-?\d+(?:\.\d+)?)", "delta"), (r"\bтеплопроводность\s*=?\s*(-?\d+(?:\.\d+)?)", "lambda_value"), (r"\bдельта t\s*=?\s*(-?\d+(?:\.\d+)?)", "delta_t"), (r"\bразность температур\s*=?\s*(-?\d+(?:\.\d+)?)", "delta_t"), (r"\bтребуемое сопротивление\s*=?\s*(-?\d+(?:\.\d+)?)", "R_tr"), (r"\bмощность\s*=?\s*(-?\d+(?:\.\d+)?)", "Q"), ] for pattern, name in extra_patterns: if name not in values: m = re.search(pattern, normalized, re.IGNORECASE) if m: values[name] = float(m.group(1)) return values def _extract_city_from_text(self, text: str) -> str | None: extracted = extract_city(text) if extracted: return self._normalize_city_name(extracted) normalized = self._normalize_text(text) for alias, canonical in sorted( self.city_aliases.items(), key=lambda item: len(item[0]), reverse=True, ): if alias and re.search(rf"\b{re.escape(alias)}\b", normalized): return canonical m = re.search( r"\b(?:для|по|в|город|города)\s+([а-яa-z\- ]{2,50})\b", normalized, ) if m: return self._normalize_city_name(m.group(1)) return None def _normalize_city_name(self, city: str | None) -> str | None: if not city: return None key = self._normalize_text(city) if key in self.city_aliases: return self.city_aliases[key] for known_city in self.cities.values(): if self._normalize_text(known_city.name) == key: return known_city.name words = [w.capitalize() for w in key.split() if w] return " ".join(words) if words else None def _normalize_material_name(self, name: str | None) -> str | None: if not name: return None key = self._normalize_text(name) if key in self.material_aliases: return self.material_aliases[key] for material in self.materials.values(): if self._normalize_text(material.name) == key: return material.name return key def _detect_formula_key( self, query: str, params: dict[str, float] | None = None, ) -> str | None: params = params or {} q = self._normalize_text(query) # Предмет расчёта важнее случайного полного набора параметров для # другой формулы. Условия после ':'/'при' не переопределяют вопрос. subject = re.split(r"[:;=]|\bпри\b|\bесли\b", q, maxsplit=1)[0] subject_patterns = { "ventilation_heat": ( r"(?:теплов[а-я]*\s+)?мощност[а-я]*\s+(?:приточн[а-я]*\s+)?вентиляц" r"|вентиляц|нагрев.{0,20}воздух|теплот.{0,30}воздух" ), "gsop": r"гсоп|градусо.?сут", "heat_loss": r"теплопотер|потер[а-я]*\s+тепл", "thermal_resistance_layer": r"(?:термическ[а-я]*\s+)?сопротивлен", "required_insulation_thickness": r"толщин[а-я]*\s+(?:слоя\s+)?(?:утепл|изоляц|теплоизоляц)", "pipe_surface_heat_flux": r"удельн[а-я]*\s+(?:теплов[а-я]*\s+)?поток|теплопоток|тепловой поток трубы", } targets = [] for key, pattern in subject_patterns.items(): matched = re.search(pattern, subject) if matched: targets.append((matched.start(), key)) for key, meta in self.formulas.items(): for alias in meta.get("aliases", []): matched = re.search(rf"(? str: # Оставляем предмет запроса, убирая только слова команды и # обычные уточняющие прилагательные, а не произвольные существительные. return re.sub( r"\b(?:пожалуйста|мне|помоги(?:те)?|помочь|можешь|можете|" r"рассчита(?:й(?:те)?|ть)|посчита(?:й(?:те)?|ть)|" r"вычисли(?:те|ть)?|определи(?:те|ть)?|найди(?:те)?|расчет|" r"требуем[а-я]*|необходим[а-я]*|суммарн[а-я]*|общ[а-я]*)\b", "", value, ).strip(" ,.-") if targets: position, key = min(targets) if without_request_words(subject[:position]): # Например, «диаметр вентиляции» не означает расчёт теплоты. return None if key == "heat_loss" and re.search(subject_patterns["ventilation_heat"], subject): return "ventilation_heat" return key remaining_subject = without_request_words(subject) # В запросе без названного предмета допускается однозначный выбор по # полному набору параметров. Названный неизвестный предмет не подменяем. if remaining_subject and remaining_subject not in { self._normalize_text(name) for name in params }: return None complete = [ key for key, meta in self.formulas.items() if set(meta["required_params"]).issubset(params) ] return complete[0] if len(complete) == 1 else None @staticmethod def _get_input_issues( query: str, formula_key: str, params: dict[str, float], ) -> list[str]: length_kind = { "pipe_surface_heat_flux": "length", "ventilation_heat": "air_flow", }.get(formula_key) issues = get_parameter_issues(query, length_kind) if {"delta_t", "t_v", "t_n"}.issubset(params): difference = params["t_v"] - params["t_n"] if not math.isclose(params["delta_t"], difference, rel_tol=1e-9, abs_tol=1e-9): issues.append( f"Задано Δt={params['delta_t']:g} °C, но t_в − t_н = " f"{params['t_v']:g} − ({params['t_n']:g}) = {difference:g} °C. " "Уточните разность температур или исходные температуры." ) return issues def _get_missing_params( self, formula_key: str, params: dict[str, float], ) -> list[str]: required = self.formulas[formula_key]["required_params"] return [name for name in required if name not in params] def can_calculate_directly( self, query: str, parameters: dict[str, Any] | None = None, ) -> bool: extracted = self._extract_parameters_from_text(query) params = self._merge_parameters(extracted, parameters or {}) formula_key = self._detect_formula_key(query, params) if formula_key is None: return False return not self._get_missing_params(formula_key, params) async def answer_calculation( self, query: str, parameters: dict[str, Any] | None = None, entities: dict[str, Any] | None = None, ) -> dict[str, Any]: self.reasoning_steps = [] provided_params = dict(parameters or {}) entities = dict(entities or {}) extracted_params = self._extract_parameters_from_text(query) params = self._merge_parameters(extracted_params, provided_params) self.reasoning_steps.append("Параметры извлечены и нормализованы") self.reasoning_steps.append(f"Параметры: {params}") formula_key = self._detect_formula_key(query, params) if formula_key is None: return self._error( "Не удалось определить тип расчёта.\n\n" "Поддерживаются: ГСОП, вентиляция, теплопотери, " "сопротивление слоя, толщина утеплителя и удельный тепловой поток." ) meta = self.formulas[formula_key] self.reasoning_steps.append(f"Выбрана формула: {meta['name']}") issues = self._get_input_issues(query, formula_key, params) if issues: return self._error("Неоднозначные или несовместимые исходные данные:\n\n" + "\n".join(issues)) missing = self._get_missing_params(formula_key, params) if not missing: self.reasoning_steps.append("Все обязательные параметры переданы") return self._execute_formula(formula_key, params) city = entities.get("city") or self._extract_city_from_text(query) if city: self.reasoning_steps.append(f"Определён город: {city}") material_name = self._extract_material_name(query) if material_name: self.reasoning_steps.append(f"Определён материал: {material_name}") params = self._try_enrich_from_material(formula_key, params, material_name) missing = self._get_missing_params(formula_key, params) if not missing: self.reasoning_steps.append("Недостающие параметры добраны из материала") return self._execute_formula(formula_key, params) climate_params = set(meta.get("climate_params", [])) missing_set = set(missing) can_use_climate_table = ( city is not None and formula_key in self.CLIMATE_DEPENDENT and missing_set and missing_set.issubset(climate_params) ) if can_use_climate_table: self.reasoning_steps.append( f"Не хватает климатических параметров {missing}; " f"используется таблица для города «{city}»" ) table_result = await self._try_table_calculation( formula_key=formula_key, city=city, params=params, ) if table_result is not None: table_result.setdefault("reasoning", "\n".join(self.reasoning_steps)) return table_result self.reasoning_steps.append(f"Недостающие параметры: {missing}") return self._missing_params(meta, missing) def _execute_formula( self, formula_key: str, params: dict[str, float], ) -> dict[str, Any]: meta = self.formulas[formula_key] handler = meta["handler"] try: for name in meta["required_params"]: if not math.isfinite(params[name]): raise ValueError(f"Параметр {name} должен быть конечным числом") result = handler(params, meta) result["reasoning"] = "\n".join(self.reasoning_steps) return self._format_result(result, meta) except (ArithmeticError, KeyError, TypeError, ValueError) as exc: logger.exception("Ошибка выполнения формулы %s", formula_key) return self._error(f"Ошибка расчёта: {exc}") async def _try_table_calculation( self, formula_key: str, city: str, params: dict[str, float], ) -> dict[str, Any] | None: if self.qa_system is None or TableCalculator is None: return None if self._table_calculator is None: self._table_calculator = TableCalculator(self.qa_system) try: # Добираем лишь недостающие значения, сохраняя заданные пользователем. # Не подставляем 20 °C молча: для теплопотерь нужны обе температуры. missing = self._get_missing_params(formula_key, params) required_climate = list(missing) if formula_key == "heat_loss" and "delta_t" in required_climate: if "t_v" not in params: return self._missing_params(self.formulas[formula_key], ["t_v", "t_n"]) required_climate = ["t_n"] climate_reader = getattr(self._table_calculator, "_climate_values_from_tables", None) if callable(climate_reader): climate = climate_reader(city, tuple(required_climate)) if climate: climate_values, source = climate enriched = self._merge_parameters(climate_values, params) if not self._get_missing_params(formula_key, enriched): self.reasoning_steps.append(f"Из таблицы для {city} получены: {climate_values}") calculated = self._execute_formula(formula_key, enriched) if not calculated.get("needs_clarification"): calculated["sources"] = [source] calculated["grounded"] = True calculated["answer"] += f"\n\nИсточник климатических данных: {source.get('doc_name', 'таблица')}" return calculated missing_reader = getattr(self._table_calculator, "_missing_climate_response", None) if callable(missing_reader): return missing_reader(city, tuple(required_climate)) return None # Совместимость с прежним TableCalculator без доступа к значениям. if formula_key == "gsop": result = self._table_calculator.calculate_gsop_from_table( city=city, t_v=params.get("t_v", 20.0), ) elif formula_key == "ventilation_heat": air_flow = params.get("L") if air_flow is None: return None result = self._table_calculator.calculate_ventilation_from_table( city=city, air_flow=air_flow, t_v=params.get("t_v", 20.0), ) elif formula_key == "heat_loss": area = params.get("A") resistance = params.get("R") if area is None or resistance is None: return None result = self._table_calculator.calculate_heat_loss_from_table( city=city, area=area, resistance=resistance, t_v=params.get("t_v", 20.0), ) else: return None if not result: return None if result.get("needs_clarification"): return result if result.get("confidence", 0.0) < 0.5: return None self.reasoning_steps.append("Табличные климатические данные успешно получены") result["reasoning"] = "\n".join(self.reasoning_steps) result.setdefault("query_type", "calculation") result.setdefault("needs_clarification", False) result.setdefault("questions", []) return result except (ArithmeticError, AttributeError, KeyError, TypeError, ValueError) as exc: logger.exception("Табличный метод расчёта недоступен") self.reasoning_steps.append(f"Табличный метод недоступен: {exc}") return None def _extract_material_name(self, text: str) -> str | None: normalized = self._normalize_text(text) for alias, canonical in sorted( self.material_aliases.items(), key=lambda item: len(item[0]), reverse=True, ): if alias and re.search(rf"\b{re.escape(alias)}\b", normalized): return canonical return None def _try_enrich_from_material( self, formula_key: str, params: dict[str, float], material_name: str | None, ) -> dict[str, float]: if not material_name: return params material = self.get_material(material_name) if material is None: return params updated = dict(params) if formula_key in {"thermal_resistance_layer", "required_insulation_thickness"}: if "lambda_value" not in updated: updated["lambda_value"] = material.lambda_value self.reasoning_steps.append( f"Теплопроводность λ взята из данных материала: {material.lambda_value}" ) return updated def add_material( self, name: str, lambda_value: float, density: float = 0.0, specific_heat: float | None = None, source: str = "", ) -> None: key = self._normalize_text(name) material = Material( name=name, lambda_value=float(lambda_value), density=float(density), specific_heat=specific_heat, source=source, ) self.materials[key] = material self._material_cache[key] = material self.material_aliases[key] = name self._save_cache() def add_city( self, name: str, t_ot: float, z_ot: int, t_n: float | None = None, source: str = "", ) -> None: key = self._normalize_text(name) city = CityClimate( name=name, t_ot=float(t_ot), z_ot=int(z_ot), t_n=float(t_n if t_n is not None else t_ot), source=source, ) self.cities[key] = city self._city_cache[key] = city self.city_aliases[key] = name self.city_list = sorted(self.cities.keys(), key=len, reverse=True) self._save_cache() def get_material(self, name: str | None) -> Material | None: canonical = self._normalize_material_name(name) if not canonical: return None key = self._normalize_text(canonical) if key in self.materials: return self.materials[key] if self.on_material_not_found is not None: try: self.on_material_not_found(canonical) except (KeyError, TypeError, ValueError, AttributeError, OSError, json.JSONDecodeError): logger.exception("Callback поиска материала завершился ошибкой") return None def get_city_climate(self, name: str | None) -> CityClimate | None: canonical = self._normalize_city_name(name) if not canonical: return None key = self._normalize_text(canonical) if key in self.cities: return self.cities[key] if self.on_city_not_found is not None: try: self.on_city_not_found(canonical) except Exception: logger.exception("Callback поиска города завершился ошибкой") return None def find_closest_city(self, query_city: str | None) -> str | None: if not query_city: return None normalized = self._normalize_text(query_city) if normalized in self.city_aliases: return self.city_aliases[normalized] for alias, canonical in self.city_aliases.items(): if normalized in alias or alias in normalized: return canonical return None @staticmethod def _calc_gsop( p: dict[str, float], meta: dict[str, Any], ) -> dict[str, Any]: t_v = p["t_v"] t_ot = p["t_ot"] z_ot = p["z_ot"] if not 0 < z_ot <= 366: raise ValueError("z_ot должно быть больше 0 и не больше 366 суток") if t_v < t_ot: raise ValueError("Для отопления t_v должна быть не ниже t_ot") value = (t_v - t_ot) * z_ot return { "answer": ( "### ГСОП\n\n" f"Подстановка: ({t_v:g} - {t_ot:g}) × {z_ot:g}\n\n" f"**Результат: {value:.0f} {meta['unit']}**" ), "result": value, "params": p, "source": meta["source"], } @staticmethod def _calc_ventilation_heat( p: dict[str, float], meta: dict[str, Any], ) -> dict[str, Any]: air_flow = p["L"] t_v = p["t_v"] t_n = p["t_n"] if air_flow < 0: raise ValueError("Расход воздуха L не может быть отрицательным") if t_v < t_n: raise ValueError("Формула нагрева требует t_v ≥ t_n; для охлаждения нужен другой расчёт") value = 0.335 * air_flow * (t_v - t_n) return { "answer": ( "### Расход теплоты на вентиляцию\n\n" f"Подстановка: Q_в = 0.335 × {air_flow:g} × ({t_v:g} - {t_n:g})\n\n" f"**Результат: {value:.1f} Вт**, или **{value / 1000:.3f} кВт**" ), "result": value, "params": p, "source": meta["source"], } @staticmethod def _calc_heat_loss( p: dict[str, float], meta: dict[str, Any], ) -> dict[str, Any]: area = p["A"] delta_t = p["delta_t"] resistance = p["R"] if area <= 0: raise ValueError("Площадь A должна быть больше 0") if resistance <= 0: raise ValueError("Сопротивление R должно быть больше 0") if delta_t < 0: raise ValueError("Для расчёта теплопотерь Δt должна быть неотрицательной") value = (area * delta_t) / resistance return { "answer": ( "### Теплопотери через ограждение\n\n" f"Подстановка: Q = ({area:g} × {delta_t:g}) / {resistance:g}\n\n" f"**Результат: {value:.1f} {meta['unit']}**, " f"или **{value / 1000:.3f} кВт**" ), "result": value, "params": p, "source": meta["source"], } @staticmethod def _calc_thermal_resistance( p: dict[str, float], meta: dict[str, Any], ) -> dict[str, Any]: delta = p["delta"] lambda_value = p["lambda_value"] if delta <= 0: raise ValueError("Толщина δ должна быть больше 0") if lambda_value <= 0: raise ValueError("Теплопроводность λ должна быть больше 0") value = delta / lambda_value return { "answer": ( "### Сопротивление слоя\n\n" f"Подстановка: R = {delta:g} / {lambda_value:g}\n\n" f"**Результат: {value:.4f} {meta['unit']}**" ), "result": value, "params": p, "source": meta["source"], } @staticmethod def _calc_insulation_thickness( p: dict[str, float], meta: dict[str, Any], ) -> dict[str, Any]: r_required = p["R_tr"] lambda_value = p["lambda_value"] if r_required <= 0: raise ValueError("R_tr должно быть больше 0") if lambda_value <= 0: raise ValueError("Теплопроводность λ должна быть больше 0") value = r_required * lambda_value return { "answer": ( "### Требуемая толщина изоляции\n\n" f"Подстановка: δ = {r_required:g} × {lambda_value:g}\n\n" f"**Результат: {value:.4f} м**, или **{value * 1000:.0f} мм**" ), "result": value, "params": p, "source": meta["source"], } @staticmethod def _calc_pipe_heat_flux( p: dict[str, float], meta: dict[str, Any], ) -> dict[str, Any]: heat_power = p["Q"] length = p["L"] if length <= 0: raise ValueError("Длина L должна быть больше 0") if heat_power < 0: raise ValueError("Тепловая мощность Q не может быть отрицательной") value = heat_power / length return { "answer": ( "### Удельный тепловой поток\n\n" f"Подстановка: q = {heat_power:g} / {length:g}\n\n" f"**Результат: {value:.2f} {meta['unit']}**" ), "result": value, "params": p, "source": meta["source"], } def _missing_params( self, meta: dict[str, Any], missing: list[str], ) -> dict[str, Any]: human_missing = [self.PARAM_LABELS.get(name, name) for name in missing] return { "answer": ( f"⚠️ Недостаточно данных для расчёта «{meta['name']}».\n\n" "Укажите:\n- " + "\n- ".join(human_missing) + f"\n\nФормула: {meta['expression']}" ), "sources": [], "tables": [], "formulas": [{ "raw": meta["expression"], "name": meta["name"], "source": meta["source"], }], "formula": { "raw": meta["expression"], "name": meta["name"], "source": meta["source"], }, "confidence": 0.2, "needs_clarification": True, "questions": [f"Укажите: {', '.join(human_missing)}"], "query_type": "calculation", } @staticmethod def _format_result( result: dict[str, Any], meta: dict[str, Any], ) -> dict[str, Any]: formula = { "raw": meta["expression"], "name": meta["name"], "source": meta["source"], } params = result.get("params", {}) legend = meta.get("legend", {}) input_lines = [ f"- {name} = {params[name]:g} — {legend.get(name, name)}" for name in meta["required_params"] if name in params ] answer = ( "### Исходные данные\n\n" + "\n".join(input_lines) + f"\n\n### Формула\n\n{meta['expression']}\n\n" + result.get("answer", "") ) if meta["id"] == "ventilation_heat": answer += "\n\nПринято постоянное значение коэффициента 0,335 Вт·ч/(м³·°C); рекуперация и влажность не учитываются." elif meta["id"] == "required_insulation_thickness": answer += "\n\nЭто оценка для одного однородного слоя; сопротивления остальных слоёв и поверхностей не учтены." return { "answer": answer, # Встроенная формула не является извлечённой цитатой из СП. "sources": [], "tables": result.get("tables", []), "formulas": [formula], "formula": formula, "params": result.get("params", {}), "result": result.get("result"), "reasoning": result.get("reasoning", ""), "confidence": result.get("confidence", 0.95), "needs_clarification": False, "questions": [], "query_type": "calculation", "grounded": False, } @staticmethod def _error(message: str) -> dict[str, Any]: return { "answer": f"❌ {message}", "sources": [], "tables": [], "formulas": [], "confidence": 0.0, "needs_clarification": True, "questions": ["Уточните запрос и исходные данные."], "query_type": "calculation", } def _build_error_response(self, message: str) -> dict[str, Any]: return self._error(message) def _build_missing_params_response( self, formula_meta: dict[str, Any], missing: list[str], ) -> dict[str, Any]: return self._missing_params(formula_meta, missing) def get_available_formulas(self) -> list[dict[str, Any]]: return [ { "id": key, "name": meta["name"], "expression": meta["expression"], "description": meta.get("description", ""), "required_params": meta["required_params"], "climate_params": meta.get("climate_params", []), "source": meta["source"], "unit": meta["unit"], "aliases": meta.get("aliases", []), "legend": meta.get("legend", {}), } for key, meta in self.formulas.items() ] def get_formula(self, formula_key: str) -> dict[str, Any] | None: return self.formulas.get(formula_key) def get_reasoning_chain(self) -> str: return "\n".join(self.reasoning_steps) def _get_reasoning_chain(self) -> str: return self.get_reasoning_chain() def explain_formula(self, formula_key: str) -> dict[str, Any]: meta = self.formulas.get(formula_key) if not meta: return self._error(f"Формула «{formula_key}» не найдена.") legend_lines = [] for key, value in meta.get("legend", {}).items(): legend_lines.append(f"- {key}: {value}") answer = ( f"### {meta['name']}\n\n" f"Формула: {meta['expression']}\n\n" f"{meta.get('description', '')}\n\n" f"{chr(10).join(legend_lines)}\n\n" f"Источник: {meta['source']}" ) return { "answer": answer, "sources": [{"doc_name": meta["source"]}], "tables": [], "formulas": [{ "raw": meta["expression"], "name": meta["name"], "source": meta["source"], }], "formula": { "raw": meta["expression"], "name": meta["name"], "source": meta["source"], }, "confidence": 0.95, "needs_clarification": False, "questions": [], "query_type": "formula_info", } def try_calculate( self, query: str, parameters: dict[str, Any] | None = None, entities: dict[str, Any] | None = None, ) -> dict[str, Any]: """ Синхронная совместимость со старым кодом. Если проект вызывает FormulaEngine без await, можно использовать этот метод. """ extracted = self._extract_parameters_from_text(query) params = self._merge_parameters(extracted, parameters or {}) formula_key = self._detect_formula_key(query, params) if formula_key is None: return self._error("Не удалось определить тип расчёта.") issues = self._get_input_issues(query, formula_key, params) if issues: return self._error("Неоднозначные или несовместимые исходные данные:\n\n" + "\n".join(issues)) meta = self.formulas[formula_key] missing = self._get_missing_params(formula_key, params) if missing: return self._missing_params(meta, missing) return self._execute_formula(formula_key, params) def try_table_calculation( self, query: str, parameters: dict[str, Any] | None = None, entities: dict[str, Any] | None = None, ) -> dict[str, Any] | None: """ Совместимость со старым API. Только определяет возможность табличного пути, но не async. Полезно для старых веток AgentLoop. """ provided_params = dict(parameters or {}) entities = dict(entities or {}) extracted_params = self._extract_parameters_from_text(query) params = self._merge_parameters(extracted_params, provided_params) formula_key = self._detect_formula_key(query, params) if formula_key is None: return None if self._get_input_issues(query, formula_key, params): return None meta = self.formulas[formula_key] missing = self._get_missing_params(formula_key, params) city = entities.get("city") or self._extract_city_from_text(query) climate_params = set(meta.get("climate_params", [])) missing_set = set(missing) if ( city is None or formula_key not in self.CLIMATE_DEPENDENT or not missing_set or not missing_set.issubset(climate_params) ): return None return { "formula_key": formula_key, "city": city, "params": params, "missing": missing, "query_type": "calculation", "can_use_table": True, } def preload_default_reference_data(self) -> None: """ Можно вызвать при старте приложения, если хочешь заранее добавить базовые материалы/города в кэш. """ defaults_materials = [ ("минеральная вата", 0.040, 50, None, "default"), ("экструдированный пенополистирол", 0.032, 35, None, "default"), ("пенополистирол", 0.038, 25, None, "default"), ("кирпич", 0.700, 1800, None, "default"), ("бетон", 1.690, 2400, None, "default"), ("газобетон", 0.120, 500, None, "default"), ] for name, lam, density, cp, source in defaults_materials: key = self._normalize_text(name) if key not in self.materials: self.add_material(name, lam, density, cp, source) defaults_cities = [ ("Москва", -3.1, 214, -3.1, "default"), ("Санкт-Петербург", -1.3, 219, -1.3, "default"), ("Екатеринбург", -6.0, 230, -6.0, "default"), ("Новосибирск", -8.7, 230, -8.7, "default"), ] for name, t_ot, z_ot, t_n, source in defaults_cities: key = self._normalize_text(name) if key not in self.cities: self.add_city(name, t_ot, z_ot, t_n, source) def debug_snapshot(self) -> dict[str, Any]: return { "materials_count": len(self.materials), "cities_count": len(self.cities), "formula_count": len(self.formulas), "city_aliases_count": len(self.city_aliases), "material_aliases_count": len(self.material_aliases), "qa_system_attached": self.qa_system is not None, "table_calculator_ready": self._table_calculator is not None, "reasoning_steps": list(self.reasoning_steps), } def validate_formula_contracts(self) -> list[str]: errors: list[str] = [] for key, meta in self.formulas.items(): for field_name in ( "id", "name", "expression", "source", "unit", "required_params", "aliases", ): if field_name not in meta: errors.append(f"{key}: missing field {field_name}") if "handler" not in meta or not callable(meta["handler"]): errors.append(f"{key}: invalid handler") if not isinstance(meta.get("required_params", []), list): errors.append(f"{key}: required_params must be list") return errors def search_materials(self, query: str) -> list[dict[str, Any]]: normalized = self._normalize_text(query) result: list[dict[str, Any]] = [] for material in self.materials.values(): name_norm = self._normalize_text(material.name) if normalized in name_norm or name_norm in normalized: result.append({ "name": material.name, "lambda_value": material.lambda_value, "density": material.density, "specific_heat": material.specific_heat, "source": material.source, }) return result[:10] def search_cities(self, query: str) -> list[dict[str, Any]]: normalized = self._normalize_text(query) result: list[dict[str, Any]] = [] for city in self.cities.values(): name_norm = self._normalize_text(city.name) if normalized in name_norm or name_norm in normalized: result.append({ "name": city.name, "t_ot": city.t_ot, "z_ot": city.z_ot, "t_n": city.t_n, "source": city.source, }) return result[:10] def _run_self_tests() -> None: """Контрольные числовые примеры и ошибки входных данных.""" import asyncio engine = FormulaEngine() examples = ( ("Рассчитай вентиляцию: L=100 м3/ч, tв=20 °C, tн=-25 °C", 1507.5), ("Рассчитай теплопотери: A=12, R=3, tв=20, tн=-30", 200.0), ("Сопротивление слоя: толщина 100 мм, λ=0,04", 2.5), ("Рассчитай толщину утеплителя: R_tr=3, λ=0,04", 0.12), ("Удельный тепловой поток: Q=2 кВт, длина трубы 10 м", 200.0), ("Рассчитай ГСОП tв=20 tот=-8,4 zот=225", 6390.0), ) for question, expected in examples: test_result = asyncio.run(engine.answer_calculation(question)) assert not test_result["needs_clarification"], test_result assert math.isclose(float(test_result["result"]), expected), test_result assert "Исходные данные" in test_result["answer"] and "Формула" in test_result["answer"] invalid = asyncio.run(engine.answer_calculation("Теплопотери A=10 R=0 dt=20")) assert invalid["needs_clarification"] and "больше 0" in invalid["answer"] overridden = asyncio.run(engine.answer_calculation("Теплопотери A=10 R=2 tв=20 tн=-20", {"t_n": -30})) assert overridden["result"] == 250.0 if __name__ == "__main__": _run_self_tests() # ИСПРАВЛЕНО: неподдерживаемый предмет не подменяется; проверяются единицы/L и конфликт Δt после API overrides; async API сохранён.