import logging import re from typing import Dict, List, Tuple, Optional logger = logging.getLogger(__name__) class CalorieDriftValidator: TOLERANCE = 0.05 # ±5% tolerance @staticmethod def create_validator() -> 'CalorieDriftValidator': return CalorieDriftValidator() def extract_meal_calories(self, meal_plan_text: str) -> List[Tuple[str, float]]: meals = [] pattern = r'([\w\s]+)\s*\(Approx\.\s*(\d+(?:\.\d+)?)\s*kcal\)' matches = re.findall(pattern, meal_plan_text, re.IGNORECASE) for meal_name, cal_value in matches: try: calories = float(cal_value) meals.append((meal_name.strip(), calories)) except ValueError: logger.warning(f"Could not parse calorie value: {cal_value}") continue logger.info(f"Extracted {len(meals)} meals with calorie values") return meals def validate_total_calories( self, meal_calories: List[Tuple[str, float]], daily_target: float ) -> Dict[str, any]: if not meal_calories: return { "valid": False, "total_extracted": 0.0, "target": daily_target, "deviation_pct": 100.0, "within_tolerance": False, "message": "No meal calories extracted" } total_extracted = sum(cal for _, cal in meal_calories) deviation = abs(total_extracted - daily_target) / daily_target deviation_pct = deviation * 100 within_tolerance = deviation <= self.TOLERANCE result = { "valid": True, "total_extracted": total_extracted, "target": daily_target, "deviation_pct": round(deviation_pct, 2), "within_tolerance": within_tolerance } if within_tolerance: logger.info( f"Calorie validation PASSED: {total_extracted} kcal " f"vs target {daily_target} kcal (deviation: {deviation_pct:.2f}%)" ) else: logger.warning( f"Calorie validation FAILED: {total_extracted} kcal " f"vs target {daily_target} kcal (deviation: {deviation_pct:.2f}% > {self.TOLERANCE*100}%)" ) return result def scale_meal_calories( self, meal_plan_text: str, scaling_factor: float ) -> str: def replace_calorie(match): meal_name = match.group(1) original_cal = float(match.group(2)) scaled_cal = round(original_cal * scaling_factor, 1) return f"{meal_name} (Approx. {int(scaled_cal)} kcal)" pattern = r'([\w\s]+)\s*\(Approx\.\s*(\d+(?:\.\d+)?)\s*kcal\)' scaled_text = re.sub(pattern, replace_calorie, meal_plan_text, flags=re.IGNORECASE) logger.info(f"Scaled meal calories by factor {scaling_factor:.3f}") return scaled_text def validate_and_correct( self, meal_plan_text: str, daily_calorie_target: float ) -> Tuple[str, Dict[str, any]]: # Extract meals meal_calories = self.extract_meal_calories(meal_plan_text) # Validate validation = self.validate_total_calories(meal_calories, daily_calorie_target) # If within tolerance, return as-is if validation["within_tolerance"] or not validation["valid"]: return meal_plan_text, validation # Calculate scaling factor total_extracted = validation["total_extracted"] scaling_factor = daily_calorie_target / total_extracted logger.info( f"Applying calorie correction: scaling by {scaling_factor:.3f} " f"to bring {total_extracted} → {daily_calorie_target} kcal" ) # Scale calories corrected_plan = self.scale_meal_calories(meal_plan_text, scaling_factor) # Re-validate corrected_meals = self.extract_meal_calories(corrected_plan) final_validation = self.validate_total_calories(corrected_meals, daily_calorie_target) final_validation["corrected"] = True final_validation["scaling_factor"] = round(scaling_factor, 3) return corrected_plan, final_validation