"""Веса строки рейса в плане на день с учётом замены компонента и правки нормы.""" from __future__ import annotations from typing import Any, Dict, Optional from app.models import Component, Ingredient, Recipe _EPSILON = 1e-6 def ingredient_dry_matter_percent(ing: Ingredient, components_by_id: Dict[str, Component]) -> float: dm = float(ing.dry_matter or 0) if dm > 0: return dm if ing.component_id and ing.component_id in components_by_id: return float(components_by_id[ing.component_id].dry_matter or 0) return 0.0 def ingredient_dry_matter_per_head( ing: Ingredient, *, dry_matter_percent: Optional[float] = None, ) -> float: if ing.dry_matter_per_head is not None and float(ing.dry_matter_per_head) > 0: return float(ing.dry_matter_per_head) wph = float(ing.weight_per_head or 0) dm = dry_matter_percent if dry_matter_percent is not None else float(ing.dry_matter or 0) if wph > 0 and dm > 0: return wph * (dm / 100.0) return 0.0 def _round_wph(value: float) -> float: if 0 < value < 0.01: return round(value, 3) return round(value, 2) def _total_kg(weight_per_head: float, heads: int, fallback_amount: float) -> float: if weight_per_head > 0: return round(weight_per_head * max(heads, 0), 2) return round(float(fallback_amount or 0), 2) def _apply_component_adjustment( *, recipe: Recipe, component_id: Optional[str], components_by_id: Dict[str, Component], master_wph: float, master_dm_ph: float, master_dm_pct: float, component_adjustment: Optional[Dict[str, Optional[float]]], ) -> tuple[float, float, float, bool]: if not component_adjustment or not component_id: return master_wph, master_dm_ph, master_dm_pct, False adj_dm_pct = component_adjustment.get("dry_matter") if adj_dm_pct is not None: new_dm_pct = float(adj_dm_pct) locked = bool(getattr(recipe, "dry_matter_locked", False)) if locked: new_dm_ph = master_dm_ph if new_dm_pct > _EPSILON: new_wph = (master_dm_ph * 100.0) / new_dm_pct if new_wph < 0.01 and master_dm_ph >= 0.001: new_wph = 0.01 else: new_wph = 0.0 return _round_wph(new_wph), round(new_dm_ph, 4), new_dm_pct, True new_dm_ph = master_wph * (new_dm_pct / 100.0) if master_wph > 0 else master_dm_ph return master_wph, round(new_dm_ph, 4), new_dm_pct, True adj_wph = component_adjustment.get("weight_per_head") adj_dm_ph = component_adjustment.get("dry_matter_per_head") if adj_wph is None and adj_dm_ph is None: return master_wph, master_dm_ph, master_dm_pct, False comp = components_by_id.get(component_id) dm_pct = float(comp.dry_matter or 0) if comp else master_dm_pct locked = bool(getattr(recipe, "dry_matter_locked", False)) if locked and adj_dm_ph is not None: new_dm_ph = float(adj_dm_ph) new_wph = (new_dm_ph * 100.0) / dm_pct if dm_pct > _EPSILON else 0.0 return _round_wph(new_wph), new_dm_ph, dm_pct, True if not locked and adj_wph is not None: new_wph = float(adj_wph) new_dm_ph = new_wph * (dm_pct / 100.0) if dm_pct > 0 else 0.0 return _round_wph(new_wph), round(new_dm_ph, 4), dm_pct, True if adj_dm_ph is not None: new_dm_ph = float(adj_dm_ph) new_wph = (new_dm_ph * 100.0) / dm_pct if dm_pct > _EPSILON else master_wph return _round_wph(new_wph), new_dm_ph, dm_pct, True if adj_wph is not None: new_wph = float(adj_wph) new_dm_ph = new_wph * (dm_pct / 100.0) if dm_pct > 0 else master_dm_ph return _round_wph(new_wph), round(new_dm_ph, 4), dm_pct, True return master_wph, master_dm_ph, master_dm_pct, False def resolve_plan_ingredient_weights( ing: Ingredient, recipe: Recipe, *, heads: int, components_by_id: Dict[str, Component], replacement_component_id: Optional[str] = None, component_adjustment: Optional[Dict[str, Optional[float]]] = None, ) -> Dict[str, Any]: """ Эффективные веса для плана. Приоритет: adjustment (по component_id) → замена → базовый рецепт. Мастер-значения ingredient не изменяются. """ master_wph = float(ing.weight_per_head or 0) master_dm_pct = ingredient_dry_matter_percent(ing, components_by_id) master_dm_ph = ingredient_dry_matter_per_head(ing, dry_matter_percent=master_dm_pct) effective_wph, effective_dm_ph, effective_dm_pct, adjusted_today = _apply_component_adjustment( recipe=recipe, component_id=ing.component_id, components_by_id=components_by_id, master_wph=master_wph, master_dm_ph=master_dm_ph, master_dm_pct=master_dm_pct, component_adjustment=component_adjustment, ) replacement = ( components_by_id.get(replacement_component_id) if replacement_component_id else None ) if replacement is None: wph = effective_wph return { "weightPerHead": wph, "totalKg": _total_kg(wph, heads, float(ing.amount or 0)), "dryMatterPct": effective_dm_pct, "dryMatterPerHead": round(effective_dm_ph, 4), "originalWeightPerHead": master_wph, "originalDryMatterPerHead": round(master_dm_ph, 4), "originalDryMatterPct": master_dm_pct, "adjustedToday": adjusted_today, "replacedToday": False, } repl_dm_pct = float(replacement.dry_matter or 0) locked = bool(getattr(recipe, "dry_matter_locked", False)) if locked: new_dm_ph = effective_dm_ph if repl_dm_pct > _EPSILON: new_wph = (new_dm_ph * 100.0) / repl_dm_pct else: new_wph = 0.0 mode = "dry_matter" else: new_wph = effective_wph new_dm_ph = new_wph * (repl_dm_pct / 100.0) if repl_dm_pct > 0 else 0.0 mode = "weight" new_wph = _round_wph(new_wph) return { "weightPerHead": new_wph, "totalKg": _total_kg(new_wph, heads, float(ing.amount or 0)), "dryMatterPct": repl_dm_pct, "dryMatterPerHead": round(new_dm_ph, 4), "originalWeightPerHead": master_wph, "originalDryMatterPerHead": round(master_dm_ph, 4), "originalDryMatterPct": master_dm_pct, "recalculationMode": mode, "adjustedToday": adjusted_today, "replacedToday": True, } # Backward-compatible aliases for internal callers _ingredient_dry_matter_percent = ingredient_dry_matter_percent _ingredient_dry_matter_per_head = ingredient_dry_matter_per_head