File size: 4,493 Bytes
345991e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
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