Spaces:
Sleeping
Sleeping
| """ | |
| Parameter Standardization Utility | |
| Handles consistent naming and mapping of soil and leaf analysis parameters | |
| across all modules in the CropDrive™ AI Assistant system. | |
| """ | |
| import re | |
| from typing import Dict, List, Any, Optional | |
| class ParameterStandardizer: | |
| """Centralized parameter standardization for soil and leaf analysis data""" | |
| def __init__(self): | |
| # Standard parameter names (canonical format) | |
| self.STANDARD_SOIL_PARAMS = { | |
| 'pH': 'pH', | |
| 'nitrogen': 'N (%)', | |
| 'organic_carbon': 'Org. C (%)', | |
| 'total_phosphorus': 'Total P (mg/kg)', | |
| 'available_phosphorus': 'Avail P (mg/kg)', | |
| 'exchangeable_potassium': 'Exch. K (meq/100 g)', | |
| 'exchangeable_calcium': 'Exch. Ca (meq/100 g)', | |
| 'exchangeable_magnesium': 'Exch. Mg (meq/100 g)', | |
| 'cec': 'CEC (meq/100 g)' | |
| } | |
| self.STANDARD_LEAF_PARAMS = { | |
| 'nitrogen': 'N (%)', | |
| 'phosphorus': 'P (%)', | |
| 'potassium': 'K (%)', | |
| 'magnesium': 'Mg (%)', | |
| 'calcium': 'Ca (%)', | |
| 'boron': 'B (mg/kg)', | |
| 'copper': 'Cu (mg/kg)', | |
| 'zinc': 'Zn (mg/kg)' | |
| } | |
| # Comprehensive mapping of all possible parameter name variations | |
| self.PARAMETER_VARIATIONS = { | |
| # pH variations | |
| 'pH': ['ph', 'pH', 'PH', 'p.h.', 'acidity', 'alkalinity'], | |
| # Nitrogen variations | |
| 'N (%)': [ | |
| 'nitrogen', 'n', 'n%', 'n (%)', 'n_%', 'nitrogen%', 'nitrogen (%)', | |
| 'nitrogen_%', 'total n', 'total nitrogen', 'total n (%)', 'total nitrogen (%)', | |
| 'Nitrogen (%)', 'Total Nitrogen (%)' | |
| ], | |
| # Organic Carbon variations | |
| 'Org. C (%)': [ | |
| 'organic carbon', 'organic_carbon', 'carbon', 'c', 'c%', 'c (%)', | |
| 'c_%', 'organic_carbon_%', 'org. c (%)', 'org c (%)', 'org.c (%)', | |
| 'Organic Carbon (%)', 'Org. C (%)' | |
| ], | |
| # Total Phosphorus variations | |
| 'Total P (mg/kg)': [ | |
| 'total phosphorus', 'total p', 'total_p', 'total phosphorus mg/kg', | |
| 'total_p_mg_kg', 'total p (mg/kg)', 'total_p_(mg/kg)' | |
| ], | |
| # Available Phosphorus variations | |
| 'Avail P (mg/kg)': [ | |
| 'available phosphorus', 'available p', 'available_p', 'avail p', | |
| 'avail_p', 'available phosphorus mg/kg', 'available_p_mg_kg', | |
| 'avail p (mg/kg)', 'avail_p_(mg/kg)', 'available p (mg/kg)', | |
| 'Available P (mg/kg)', 'Available Phosphorus (mg/kg)' | |
| ], | |
| # Exchangeable Potassium variations | |
| 'Exch. K (meq/100 g)': [ | |
| 'exchangeable potassium', 'exch k', 'exch_k', 'exchangeable k', | |
| 'exchangeable_k', 'k meq%', 'k_meq%', 'exchangeable_k_meq%', | |
| 'exch. k (meq%)', 'exch k (meq%)', 'exch. k meq%', 'exch k meq%', | |
| 'exch k (cmol/kg)', 'exch. k (cmol/kg)', 'exch k cmol/kg', 'exch. k cmol/kg', | |
| 'k meq/100 g', 'k_meq/100 g', 'exchangeable_k_meq/100 g', | |
| 'exch. k (meq/100 g)', 'exch k (meq/100 g)', 'exch. k meq/100 g', 'exch k meq/100 g' | |
| ], | |
| # Exchangeable Calcium variations | |
| 'Exch. Ca (meq/100 g)': [ | |
| 'exchangeable calcium', 'exch ca', 'exch_ca', 'exchangeable ca', | |
| 'exchangeable_ca', 'ca meq%', 'ca_meq%', 'exchangeable_ca_meq%', | |
| 'exch. ca (meq%)', 'exch ca (meq%)', 'exch. ca meq%', 'exch ca meq%', | |
| 'exch ca (cmol/kg)', 'exch. ca (cmol/kg)', 'exch ca cmol/kg', 'exch. ca cmol/kg', | |
| 'ca meq/100 g', 'ca_meq/100 g', 'exchangeable_ca_meq/100 g', | |
| 'exch. ca (meq/100 g)', 'exch ca (meq/100 g)', 'exch. ca meq/100 g', 'exch ca meq/100 g' | |
| ], | |
| # Exchangeable Magnesium variations | |
| 'Exch. Mg (meq/100 g)': [ | |
| 'exchangeable magnesium', 'exch mg', 'exch_mg', 'exchangeable mg', | |
| 'exchangeable_mg', 'mg meq%', 'mg_meq%', 'exchangeable_mg_meq%', | |
| 'exch. mg (meq%)', 'exch mg (meq%)', 'exch. mg meq%', 'exch mg meq%', | |
| 'exch mg (cmol/kg)', 'exch. mg (cmol/kg)', 'exch mg cmol/kg', 'exch. mg cmol/kg', | |
| 'mg meq/100 g', 'mg_meq/100 g', 'exchangeable_mg_meq/100 g', | |
| 'exch. mg (meq/100 g)', 'exch mg (meq/100 g)', 'exch. mg meq/100 g', 'exch mg meq/100 g' | |
| ], | |
| # CEC variations | |
| 'CEC (meq/100 g)': [ | |
| 'cec', 'cation exchange capacity', 'c.e.c', 'cec meq%', 'cec_meq%', | |
| 'cec (meq%)', 'c.e.c (meq%)', 'c.e.c meq%', 'cec (cmol/kg)', 'cec cmol/kg', | |
| 'C.E.C (meq%)', 'CEC (meq%)', 'cec meq/100 g', 'cec_meq/100 g', | |
| 'cec (meq/100 g)', 'c.e.c (meq/100 g)', 'c.e.c meq/100 g', | |
| 'C.E.C (meq/100 g)', 'CEC (meq/100 g)' | |
| ], | |
| # Leaf parameter variations | |
| 'P (%)': [ | |
| 'phosphorus', 'p', 'p%', 'p (%)', 'p_%', 'phosphorus%', 'phosphorus (%)', | |
| 'leaf phosphorus', 'leaf p', 'leaf_p' | |
| ], | |
| 'K (%)': [ | |
| 'potassium', 'k', 'k%', 'k (%)', 'k_%', 'potassium%', 'potassium (%)', | |
| 'leaf potassium', 'leaf k', 'leaf_k' | |
| ], | |
| 'Mg (%)': [ | |
| 'magnesium', 'mg', 'mg%', 'mg (%)', 'mg_%', 'magnesium%', 'magnesium (%)', | |
| 'leaf magnesium', 'leaf mg', 'leaf_mg' | |
| ], | |
| 'Ca (%)': [ | |
| 'calcium', 'ca', 'ca%', 'ca (%)', 'ca_%', 'calcium%', 'calcium (%)', | |
| 'leaf calcium', 'leaf ca', 'leaf_ca' | |
| ], | |
| 'B (mg/kg)': [ | |
| 'boron', 'b', 'b mg/kg', 'b_mg_kg', 'boron mg/kg', 'boron_mg_kg', | |
| 'b (mg/kg)', 'boron (mg/kg)', 'leaf boron', 'leaf b', 'leaf_b' | |
| ], | |
| 'Cu (mg/kg)': [ | |
| 'copper', 'cu', 'cu mg/kg', 'cu_mg_kg', 'copper mg/kg', 'copper_mg_kg', | |
| 'cu (mg/kg)', 'copper (mg/kg)', 'leaf copper', 'leaf cu', 'leaf_cu' | |
| ], | |
| 'Zn (mg/kg)': [ | |
| 'zinc', 'zn', 'zn mg/kg', 'zn_mg_kg', 'zinc mg/kg', 'zinc_mg_kg', | |
| 'zn (mg/kg)', 'zinc (mg/kg)', 'leaf zinc', 'leaf zn', 'leaf_zn' | |
| ] | |
| } | |
| # Create reverse mapping for quick lookup | |
| self.variation_to_standard = {} | |
| for standard, variations in self.PARAMETER_VARIATIONS.items(): | |
| for variation in variations: | |
| self.variation_to_standard[variation.lower()] = standard | |
| def standardize_parameter_name(self, param_name: str) -> Optional[str]: | |
| """ | |
| Convert any parameter name variation to the standard format | |
| Args: | |
| param_name: The parameter name to standardize | |
| Returns: | |
| Standard parameter name or None if not found | |
| """ | |
| if not param_name: | |
| return None | |
| # Clean the parameter name | |
| clean_name = param_name.strip().lower() | |
| clean_name = re.sub(r'\s+', ' ', clean_name) # Normalize whitespace | |
| # Direct lookup | |
| if clean_name in self.variation_to_standard: | |
| return self.variation_to_standard[clean_name] | |
| # Fuzzy matching for partial matches (more precise) | |
| for variation, standard in self.variation_to_standard.items(): | |
| # Only match if the variation is a significant part of the clean_name | |
| # Avoid matching single characters or very short variations | |
| if len(variation) >= 3 and (variation in clean_name or clean_name in variation): | |
| return standard | |
| return None | |
| def standardize_data_dict(self, data_dict: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Standardize all parameter names in a data dictionary | |
| Args: | |
| data_dict: Dictionary with potentially non-standard parameter names | |
| Returns: | |
| Dictionary with standardized parameter names | |
| """ | |
| standardized = {} | |
| for key, value in data_dict.items(): | |
| standard_key = self.standardize_parameter_name(key) | |
| if standard_key: | |
| standardized[standard_key] = value | |
| else: | |
| # Keep non-parameter keys as-is (like sample_id, lab_no, etc.) | |
| standardized[key] = value | |
| return standardized | |
| def standardize_samples_list(self, samples: List[Dict[str, Any]]) -> List[Dict[str, Any]]: | |
| """ | |
| Standardize parameter names in a list of sample dictionaries | |
| Args: | |
| samples: List of sample dictionaries | |
| Returns: | |
| List of samples with standardized parameter names | |
| """ | |
| return [self.standardize_data_dict(sample) for sample in samples] | |
| def get_display_name_mapping(self, param_type: str = 'soil') -> Dict[str, str]: | |
| """ | |
| Get mapping from standard parameter names to display names | |
| Args: | |
| param_type: 'soil' or 'leaf' | |
| Returns: | |
| Dictionary mapping standard names to display names | |
| """ | |
| if param_type.lower() == 'soil': | |
| return { | |
| 'pH': 'pH', | |
| 'N (%)': 'Nitrogen (%)', | |
| 'Org. C (%)': 'Organic Carbon (%)', | |
| 'Total P (mg/kg)': 'Total P (mg/kg)', | |
| 'Avail P (mg/kg)': 'Available P (mg/kg)', | |
| 'Exch. K (meq/100 g)': 'Exchangeable K (meq/100 g)', | |
| 'Exch. Ca (meq/100 g)': 'Exchangeable Ca (meq/100 g)', | |
| 'Exch. Mg (meq/100 g)': 'Exchangeable Mg (meq/100 g)', | |
| 'CEC (meq/100 g)': 'CEC (meq/100 g)' | |
| } | |
| else: # leaf | |
| return { | |
| 'N (%)': 'Nitrogen (%)', | |
| 'P (%)': 'Phosphorus (%)', | |
| 'K (%)': 'Potassium (%)', | |
| 'Mg (%)': 'Magnesium (%)', | |
| 'Ca (%)': 'Calcium (%)', | |
| 'B (mg/kg)': 'Boron (mg/kg)', | |
| 'Cu (mg/kg)': 'Copper (mg/kg)', | |
| 'Zn (mg/kg)': 'Zinc (mg/kg)' | |
| } | |
| def validate_parameter_completeness(self, data_dict: Dict[str, Any], param_type: str = 'soil') -> Dict[str, Any]: | |
| """ | |
| Validate that all expected parameters are present and add missing ones with default values | |
| Args: | |
| data_dict: Data dictionary to validate | |
| param_type: 'soil' or 'leaf' | |
| Returns: | |
| Dictionary with all expected parameters | |
| """ | |
| if param_type.lower() == 'soil': | |
| expected_params = list(self.STANDARD_SOIL_PARAMS.values()) | |
| else: | |
| expected_params = list(self.STANDARD_LEAF_PARAMS.values()) | |
| validated_data = data_dict.copy() | |
| for param in expected_params: | |
| if param not in validated_data: | |
| validated_data[param] = 0.0 # Default value for missing parameters | |
| return validated_data | |
| def get_parameter_variations_mapping(cls) -> Dict[str, List[str]]: | |
| """Get parameter variations mapping for text analysis""" | |
| return { | |
| # Soil Parameters (9) | |
| 'ph': ['ph', 'ph level', 'soil ph', 'acidity', 'alkalinity'], | |
| 'nitrogen': ['nitrogen', 'n', 'n%', 'n_%', 'nitrogen%', 'nitrogen_%', 'n (%)', 'n(%)'], | |
| 'organic_carbon': ['organic carbon', 'organic_carbon', 'carbon', 'c', 'c%', 'c_%', 'organic_carbon_%', 'org. c (%)', 'org c (%)', 'org.c(%)'], | |
| 'total_phosphorus': ['total phosphorus', 'total p', 'total_p', 'total phosphorus mg/kg', 'total_p_mg_kg', 'total p (mg/kg)', 'total p(mg/kg)'], | |
| 'available_phosphorus': ['available phosphorus', 'available p', 'available_p', 'available phosphorus mg/kg', 'available_p_mg_kg', 'avail p (mg/kg)', 'avail p(mg/kg)'], | |
| 'exchangeable_potassium': ['exchangeable potassium', 'exch k', 'exch_k', 'exchangeable k', 'exchangeable_k', 'k meq%', 'k_meq%', 'exchangeable_k_meq%', 'exch. k (meq%)', 'exch k (meq%)', 'exch.k(meq%)'], | |
| 'exchangeable_calcium': ['exchangeable calcium', 'exch ca', 'exch_ca', 'exchangeable ca', 'exchangeable_ca', 'ca meq%', 'ca_meq%', 'exchangeable_ca_meq%', 'exch. ca (meq%)', 'exch ca (meq%)', 'exch.ca(meq%)'], | |
| 'exchangeable_magnesium': ['exchangeable magnesium', 'exch mg', 'exch_mg', 'exchangeable mg', 'exchangeable_mg', 'mg meq%', 'mg_meq%', 'exchangeable_mg_meq%', 'exch. mg (meq%)', 'exch mg (meq%)', 'exch.mg(meq%)'], | |
| 'cec': ['cec', 'cation exchange capacity', 'c.e.c', 'cec meq%', 'cec_meq%', 'cec (meq%)', 'cec(meq%)'], | |
| # Leaf Parameters (8) | |
| 'leaf_nitrogen': ['leaf nitrogen', 'leaf n', 'leaf_n', 'n%', 'n_%', 'nitrogen%', 'nitrogen_%', 'n (%)', 'n(%)'], | |
| 'leaf_phosphorus': ['leaf phosphorus', 'leaf p', 'leaf_p', 'p%', 'p_%', 'phosphorus%', 'phosphorus_%', 'p (%)', 'p(%)'], | |
| 'leaf_potassium': ['leaf potassium', 'leaf k', 'leaf_k', 'k%', 'k_%', 'potassium%', 'potassium_%', 'k (%)', 'k(%)'], | |
| 'leaf_magnesium': ['leaf magnesium', 'leaf mg', 'leaf_mg', 'mg%', 'mg_%', 'magnesium%', 'magnesium_%', 'mg (%)', 'mg(%)'], | |
| 'leaf_calcium': ['leaf calcium', 'leaf ca', 'leaf_ca', 'ca%', 'ca_%', 'calcium%', 'calcium_%', 'ca (%)', 'ca(%)'], | |
| 'leaf_boron': ['leaf boron', 'leaf b', 'leaf_b', 'b mg/kg', 'b_mg_kg', 'boron mg/kg', 'boron_mg_kg', 'b (mg/kg)', 'b(mg/kg)'], | |
| 'leaf_copper': ['leaf copper', 'leaf cu', 'leaf_cu', 'cu mg/kg', 'cu_mg_kg', 'copper mg/kg', 'copper_mg_kg', 'cu (mg/kg)', 'cu(mg/kg)'], | |
| 'leaf_zinc': ['leaf zinc', 'leaf zn', 'leaf_zn', 'zn mg/kg', 'zn_mg_kg', 'zinc mg/kg', 'zinc_mg_kg', 'zn (mg/kg)', 'zn(mg/kg)'], | |
| # Land & Yield Parameters | |
| 'land_size': ['land size', 'land_size', 'farm size', 'farm_size', 'area', 'hectares', 'acres', 'square meters', 'square_meters'], | |
| 'current_yield': ['current yield', 'current_yield', 'yield', 'production', 'tonnes/hectare', 'kg/hectare', 'tonnes/acre', 'kg/acre', 'yield per hectare', 'yield per acre'], | |
| 'yield_forecast': ['yield forecast', 'yield_forecast', 'projected yield', 'projected_yield', 'future yield', 'future_yield', 'yield projection', 'yield_projection'], | |
| 'economic_impact': ['economic impact', 'economic_impact', 'roi', 'return on investment', 'cost benefit', 'cost_benefit', 'profitability', 'revenue', 'income'], | |
| # Legacy mappings for backward compatibility | |
| 'phosphorus': ['phosphorus', 'p', 'p%', 'p_%', 'phosphorus%', 'available p'], | |
| 'potassium': ['potassium', 'k', 'k%', 'k_%', 'potassium%'], | |
| 'calcium': ['calcium', 'ca', 'ca%', 'ca_%', 'calcium%'], | |
| 'magnesium': ['magnesium', 'mg', 'mg%', 'mg_%', 'magnesium%'], | |
| 'copper': ['copper', 'cu', 'cu mg/kg', 'cu_mg/kg', 'copper mg/kg'], | |
| 'zinc': ['zinc', 'zn', 'zn mg/kg', 'zn_mg/kg', 'zinc mg/kg'], | |
| 'boron': ['boron', 'b', 'b mg/kg', 'b_mg/kg', 'boron mg/kg'] | |
| } | |
| # Global instance for easy import | |
| parameter_standardizer = ParameterStandardizer() |