Spaces:
Runtime error
Runtime error
| """ | |
| OMIRL Shared Validation and Configuration | |
| This module provides configuration-based validation for OMIRL parameters. | |
| It loads validation rules and parameter definitions from YAML files, | |
| making the system more maintainable and flexible. | |
| """ | |
| import yaml | |
| from typing import List, Dict, Set, Optional, Tuple, Any | |
| from pathlib import Path | |
| import difflib | |
| class OMIRLValidator: | |
| """ | |
| Configuration-based validator for OMIRL parameters | |
| Loads validation rules and parameter definitions from YAML files, | |
| providing flexible validation with auto-correction and suggestions. | |
| """ | |
| def __init__(self, config_dir: Optional[Path] = None): | |
| if config_dir is None: | |
| # Default to config directory relative to this file | |
| config_dir = Path(__file__).parent.parent / "config" | |
| self.config_dir = Path(config_dir) | |
| self._load_configurations() | |
| def _load_configurations(self): | |
| """Load all YAML configuration files""" | |
| try: | |
| # Load parameter definitions (OMIRL-specific) | |
| with open(self.config_dir / "parameters.yaml", 'r', encoding='utf-8') as f: | |
| self.parameters = yaml.safe_load(f) | |
| # Load geographic data from global configuration | |
| self._load_geography_config() | |
| # Load mode/task combinations | |
| with open(self.config_dir / "tasks.yaml", 'r', encoding='utf-8') as f: | |
| self.mode_tasks = yaml.safe_load(f) | |
| # Load validation rules | |
| with open(self.config_dir / "validation_rules.yaml", 'r', encoding='utf-8') as f: | |
| self.validation_rules = yaml.safe_load(f) | |
| except FileNotFoundError as e: | |
| raise RuntimeError(f"OMIRL configuration file not found: {e}") | |
| except yaml.YAMLError as e: | |
| raise RuntimeError(f"Error parsing OMIRL configuration: {e}") | |
| def _load_geography_config(self): | |
| """Load geographic data from the global configuration""" | |
| try: | |
| # Look for global geography configuration | |
| global_config_path = Path(__file__).parent.parent.parent.parent / "agent" / "config" / "geography.yaml" | |
| if global_config_path.exists(): | |
| with open(global_config_path, 'r', encoding='utf-8') as f: | |
| geography_config = yaml.safe_load(f) | |
| # Extract Liguria region data | |
| liguria_data = geography_config.get("regions", {}).get("liguria", {}) | |
| # Store geography data for comune validation | |
| self._geography_data = liguria_data | |
| # Add provinces to parameters for backward compatibility | |
| if "provinces" in liguria_data: | |
| self.parameters["provinces"] = liguria_data["provinces"] | |
| # Add alert zones to parameters for backward compatibility | |
| if "alert_zones" in liguria_data: | |
| # Convert alert zones format to match expected format | |
| alert_zones = liguria_data["alert_zones"] | |
| if isinstance(alert_zones, dict) and "zones" in alert_zones: | |
| self.parameters["alert_zones"] = alert_zones["zones"] | |
| else: | |
| self.parameters["alert_zones"] = alert_zones | |
| else: | |
| raise RuntimeError(f"Global geography configuration not found at {global_config_path}") | |
| except Exception as e: | |
| raise RuntimeError(f"Error loading geography configuration: {e}") | |
| def validate_sensor_type(self, sensor_type: str) -> Tuple[bool, Optional[str], List[str]]: | |
| """ | |
| Validate sensor type parameter | |
| Returns: | |
| Tuple of (is_valid, corrected_value, suggestions) | |
| """ | |
| if not sensor_type: | |
| return True, None, [] | |
| valid_types = self.parameters['sensor_types'] | |
| # Exact match (case insensitive if configured) | |
| if self.validation_rules['rules']['sensor_type']['case_sensitive']: | |
| if sensor_type in valid_types: | |
| return True, sensor_type, [] | |
| else: | |
| # Case insensitive matching | |
| for valid_type in valid_types: | |
| if sensor_type.lower() == valid_type.lower(): | |
| return True, valid_type, [] | |
| # Generate suggestions using fuzzy matching | |
| suggestions = [] | |
| if self.validation_rules['validation_settings']['provide_suggestions']: | |
| suggestions = self._get_suggestions(sensor_type, valid_types) | |
| return False, None, suggestions | |
| def validate_provincia(self, provincia: str, task: str = None) -> Tuple[bool, Optional[str], List[str]]: | |
| """ | |
| Validate and normalize province parameter with task-specific formatting | |
| Args: | |
| provincia: Province name or code to validate | |
| task: Task name for task-specific formatting (valori_stazioni, massimi_precipitazione, etc.) | |
| Returns: | |
| Tuple of (is_valid, normalized_value, suggestions) | |
| """ | |
| if not provincia: | |
| return True, None, [] | |
| provincia_config = self.parameters['provinces'] | |
| rules = self.validation_rules['rules']['provincia'] | |
| # Determine target format based on task | |
| target_format = "codes" # Default | |
| if task and 'task_specific_formats' in rules: | |
| target_format = rules['task_specific_formats'].get(task, target_format) | |
| elif rules.get('auto_convert_to_codes', True): | |
| target_format = "codes" | |
| # For case-insensitive matching, but preserve proper format for return | |
| provincia_check = provincia if rules['case_sensitive'] else provincia.upper() | |
| # Check if it's already a valid code | |
| if rules['allow_codes'] and provincia_check in provincia_config['codes']: | |
| if target_format == "codes": | |
| return True, provincia_check, [] | |
| else: # target_format == "names" | |
| # Convert code to name (return proper case name) | |
| code_to_name = {v: k for k, v in provincia_config['name_to_code_mapping'].items()} | |
| if provincia_check in code_to_name: | |
| return True, code_to_name[provincia_check], [] | |
| # Check if it's a full name - handle case-insensitive matching | |
| if rules['allow_full_names']: | |
| # For names, we need to find the exact match regardless of case | |
| name_mapping = provincia_config['name_to_code_mapping'] | |
| # Find matching name (case-insensitive) | |
| matched_name = None | |
| for valid_name in name_mapping.keys(): | |
| if (rules['case_sensitive'] and provincia == valid_name) or \ | |
| (not rules['case_sensitive'] and provincia.upper() == valid_name.upper()): | |
| matched_name = valid_name | |
| break | |
| if matched_name: | |
| if target_format == "names": | |
| # Return the properly formatted name from configuration | |
| return True, matched_name, [] | |
| else: # target_format == "codes" | |
| code = name_mapping[matched_name] | |
| return True, code, [] | |
| # Generate suggestions | |
| suggestions = [] | |
| if self.validation_rules['validation_settings']['provide_suggestions']: | |
| all_valid = provincia_config['codes'] + provincia_config['full_names'] | |
| suggestions = self._get_suggestions(provincia, all_valid) | |
| return False, None, suggestions | |
| def validate_zona(self, zona: str) -> Tuple[bool, Optional[str], List[str]]: | |
| """Validate alert zone parameter""" | |
| if not zona: | |
| return True, None, [] | |
| valid_zones = self.parameters['alert_zones'] | |
| rules = self.validation_rules['rules']['zona'] | |
| zona_check = zona if rules['case_sensitive'] else zona.upper() | |
| if zona_check in valid_zones: | |
| return True, zona_check, [] | |
| suggestions = [] | |
| if self.validation_rules['validation_settings']['provide_suggestions']: | |
| suggestions = self._get_suggestions(zona, valid_zones) | |
| return False, None, suggestions | |
| def validate_comune(self, comune: str, provincia: str = None) -> Tuple[bool, Optional[str], Optional[str], List[str]]: | |
| """ | |
| Validate comune parameter and optionally infer province | |
| Args: | |
| comune: Municipality name to validate | |
| provincia: Optional province to validate against | |
| Returns: | |
| Tuple of (is_valid, normalized_comune, inferred_provincia, suggestions) | |
| """ | |
| if not comune: | |
| return True, None, None, [] | |
| rules = self.validation_rules['rules']['comune'] | |
| # Get municipality data from global geography (loaded via _load_geography_config) | |
| comuni_by_province = getattr(self, '_comuni_by_province', {}) | |
| if not comuni_by_province: | |
| # Extract from parameters if not cached | |
| # This would contain the loaded geography data | |
| comuni_by_province = {} | |
| # Build municipality lookup if not already built | |
| if not hasattr(self, '_municipality_to_province'): | |
| self._municipality_to_province = {} | |
| # Extract from the loaded geography data in parameters | |
| # The geography config loading should populate this | |
| pass | |
| # Normalize case for comparison | |
| comune_check = comune if rules['case_sensitive'] else comune.title() | |
| # Look for the municipality across all provinces | |
| found_province = None | |
| normalized_comune = None | |
| # Search in loaded geography data | |
| if hasattr(self, '_geography_data'): | |
| liguria_data = self._geography_data | |
| comuni_by_province = liguria_data.get('comuni_by_province', {}) | |
| for prov_name, comuni_list in comuni_by_province.items(): | |
| if isinstance(comuni_list, list): | |
| for comune_name in comuni_list: | |
| if (rules['case_sensitive'] and comune == comune_name) or \ | |
| (not rules['case_sensitive'] and comune.upper() == comune_name.upper()): | |
| found_province = prov_name | |
| normalized_comune = comune_name | |
| break | |
| if found_province: | |
| break | |
| # If provincia is specified, validate the comune exists in that province | |
| if provincia and rules.get('require_provincia_match', False): | |
| if found_province and found_province.upper() != provincia.upper(): | |
| return False, None, None, [f"Il comune '{comune}' non si trova nella provincia '{provincia}' ma in '{found_province}'"] | |
| if found_province: | |
| return True, normalized_comune, found_province, [] | |
| # Generate suggestions if not found | |
| suggestions = [] | |
| if self.validation_rules['validation_settings']['provide_suggestions']: | |
| # Collect all municipality names for suggestions | |
| all_comuni = [] | |
| if hasattr(self, '_geography_data'): | |
| liguria_data = self._geography_data | |
| comuni_by_province = liguria_data.get('comuni_by_province', {}) | |
| for comuni_list in comuni_by_province.values(): | |
| if isinstance(comuni_list, list): | |
| all_comuni.extend(comuni_list) | |
| suggestions = self._get_suggestions(comune, all_comuni) | |
| return False, None, None, suggestions | |
| def validate_periodo(self, periodo: str) -> Tuple[bool, Optional[str], List[str]]: | |
| """Validate time period parameter""" | |
| if not periodo: | |
| return True, None, [] | |
| valid_periods = self.parameters['time_periods'] | |
| rules = self.validation_rules['rules']['periodo'] | |
| periodo_check = periodo if rules['case_sensitive'] else periodo.lower() | |
| # Check exact match | |
| for valid_period in valid_periods: | |
| if periodo_check == (valid_period if rules['case_sensitive'] else valid_period.lower()): | |
| return True, valid_period, [] | |
| suggestions = [] | |
| if self.validation_rules['validation_settings']['provide_suggestions']: | |
| suggestions = self._get_suggestions(periodo, valid_periods) | |
| return False, None, suggestions | |
| # DEPRECATED: validate_mode_task_combination - mode parameter removed | |
| # def validate_mode_task_combination(self, mode: str, task: str) -> Tuple[bool, List[str]]: | |
| # """ | |
| # Validate that mode and task combination is supported | |
| # | |
| # Returns: | |
| # Tuple of (is_valid, valid_tasks_for_mode) | |
| # """ | |
| # valid_combinations = self.mode_tasks['valid_combinations'] | |
| # | |
| # if mode not in valid_combinations: | |
| # return False, [] | |
| # | |
| # valid_tasks = valid_combinations[mode] | |
| # return task in valid_tasks, valid_tasks | |
| def validate_variabile_climatica(self, variabile: str) -> Tuple[bool, Optional[str], List[str]]: | |
| """Validate climate variable parameter""" | |
| if not variabile: | |
| return True, None, [] | |
| valid_variables = self.parameters['climate_variables'] | |
| # Case insensitive matching | |
| for valid_var in valid_variables: | |
| if variabile.lower() == valid_var.lower(): | |
| return True, valid_var, [] | |
| suggestions = [] | |
| if self.validation_rules['validation_settings']['provide_suggestions']: | |
| suggestions = self._get_suggestions(variabile, valid_variables) | |
| return False, None, suggestions | |
| def validate_satellite_area(self, area: str) -> Tuple[bool, Optional[str], List[str]]: | |
| """Validate satellite area parameter""" | |
| if not area: | |
| return True, None, [] | |
| valid_areas = self.parameters['satellite_areas'] | |
| # Case insensitive matching | |
| for valid_area in valid_areas: | |
| if area.lower() == valid_area.lower(): | |
| return True, valid_area, [] | |
| suggestions = [] | |
| if self.validation_rules['validation_settings']['provide_suggestions']: | |
| suggestions = self._get_suggestions(area, valid_areas) | |
| return False, None, suggestions | |
| def get_task_url(self, mode: str, task: str) -> Optional[str]: | |
| """Get the URL for a specific mode/task combination""" | |
| task_urls = self.parameters.get('task_urls', {}) | |
| return task_urls.get(mode, {}).get(task) | |
| def get_task_requirements(self, task: str) -> Dict[str, Any]: | |
| """Get requirements for a specific task""" | |
| return self.mode_tasks.get('task_requirements', {}).get(task, {}) | |
| def validate_complete_request(self, task: str, filters: Dict[str, Any]) -> Tuple[bool, Dict[str, Any], List[str]]: | |
| """ | |
| Validate a complete OMIRL request | |
| Returns: | |
| Tuple of (is_valid, corrected_filters, error_messages) | |
| """ | |
| errors = [] | |
| corrected_filters = filters.copy() | |
| # Validate task exists (no mode validation needed) | |
| valid_tasks = self.mode_tasks.get('valid_tasks', []) | |
| if task not in valid_tasks: | |
| errors.append(f"Invalid task '{task}'. Valid tasks: {valid_tasks}") | |
| return False, corrected_filters, errors | |
| # Validate filters are allowed for this task | |
| task_requirements = self.get_task_requirements(task) | |
| allowed_filters = set(task_requirements.get('required_filters', []) + task_requirements.get('optional_filters', [])) | |
| # Remove filters that are not allowed for this task | |
| for filter_name in list(corrected_filters.keys()): | |
| if filter_name not in allowed_filters: | |
| print(f"⚠️ Removing unsupported filter '{filter_name}' for task '{task}'. Allowed filters: {list(allowed_filters)}") | |
| del corrected_filters[filter_name] | |
| # Validate individual filters | |
| validation_methods = { | |
| 'tipo_sensore': self.validate_sensor_type, | |
| 'zona': self.validate_zona, | |
| 'periodo': self.validate_periodo, | |
| 'variabile_climatica': self.validate_variabile_climatica | |
| } | |
| for filter_name, filter_value in filters.items(): | |
| if filter_name in validation_methods and filter_value: | |
| is_valid, corrected_value, suggestions = validation_methods[filter_name](filter_value) | |
| if not is_valid: | |
| error_msg = f"Invalid {filter_name}: '{filter_value}'" | |
| if suggestions: | |
| error_msg += f". Suggestions: {', '.join(suggestions[:3])}" | |
| errors.append(error_msg) | |
| elif corrected_value and corrected_value != filter_value: | |
| # Auto-correction applied | |
| corrected_filters[filter_name] = corrected_value | |
| elif filter_name == 'provincia' and filter_value: | |
| # Special handling for provincia with task-specific formatting | |
| is_valid, corrected_value, suggestions = self.validate_provincia(filter_value, task) | |
| if not is_valid: | |
| error_msg = f"Invalid provincia: '{filter_value}'" | |
| if suggestions: | |
| error_msg += f". Suggestions: {', '.join(suggestions[:3])}" | |
| errors.append(error_msg) | |
| elif corrected_value and corrected_value != filter_value: | |
| # Auto-correction applied | |
| corrected_filters[filter_name] = corrected_value | |
| return len(errors) == 0, corrected_filters, errors | |
| def _get_suggestions(self, invalid_value: str, valid_options: List[str]) -> List[str]: | |
| """Generate suggestions using fuzzy string matching""" | |
| threshold = self.validation_rules['suggestion_settings']['similarity_threshold'] | |
| max_suggestions = self.validation_rules['suggestion_settings']['max_suggestions'] | |
| suggestions = difflib.get_close_matches( | |
| invalid_value, | |
| valid_options, | |
| n=max_suggestions, | |
| cutoff=threshold | |
| ) | |
| return suggestions | |
| def get_valid_tasks(self) -> List[str]: | |
| """Get list of valid tasks""" | |
| return self.parameters.get('valid_tasks', []).copy() | |
| def get_task_default(self, task: str, parameter: str) -> Optional[str]: | |
| """Get default value for a task parameter""" | |
| defaults = self.parameters.get('defaults', {}) | |
| task_defaults = defaults.get(task, {}) | |
| return task_defaults.get(parameter) | |
| def get_task_url(self, task: str, mode: str = "tables") -> Optional[str]: | |
| """Get the URL for a specific task and mode""" | |
| task_urls = self.parameters.get('task_urls', {}) | |
| mode_urls = task_urls.get(mode, {}) | |
| return mode_urls.get(task) | |
| def get_time_periods(self) -> List[str]: | |
| """Get list of valid time periods""" | |
| return self.parameters.get('time_periods', []).copy() | |
| def get_period_mappings(self) -> Dict[str, str]: | |
| """Get period normalization mappings""" | |
| return self.parameters.get('period_mappings', {}).copy() | |
| def normalize_periodo(self, periodo: str) -> Optional[str]: | |
| """ | |
| Normalize period parameter using configuration mappings | |
| Args: | |
| periodo: Period string to normalize | |
| Returns: | |
| Normalized period string or None if invalid | |
| """ | |
| if not periodo: | |
| return None | |
| period_mappings = self.get_period_mappings() | |
| # Direct mapping | |
| if periodo in period_mappings: | |
| return period_mappings[periodo] | |
| # Case-insensitive mapping | |
| for key, value in period_mappings.items(): | |
| if periodo.lower() == key.lower(): | |
| return value | |
| # If no mapping found, check if it's already a valid standard format | |
| valid_periods = self.get_time_periods() | |
| if periodo in valid_periods: | |
| return periodo | |
| return None | |
| # Convenience methods for backward compatibility | |
| def get_valid_sensor_types(self) -> List[str]: | |
| """Get list of valid sensor types""" | |
| return self.parameters['sensor_types'].copy() | |
| def get_valid_provinces(self) -> Dict[str, str]: | |
| """Get mapping of province names to codes""" | |
| return self.parameters['provinces']['name_to_code_mapping'].copy() | |
| def get_alert_zones(self) -> List[str]: | |
| """Get list of valid alert zones from global geography configuration""" | |
| return self.parameters.get('alert_zones', []) | |
| # Global validator instance (lazy loaded) | |
| _validator_instance = None | |
| def get_validator() -> OMIRLValidator: | |
| """Get the global validator instance""" | |
| global _validator_instance | |
| if _validator_instance is None: | |
| _validator_instance = OMIRLValidator() | |
| return _validator_instance | |
| # Convenience functions for backward compatibility | |
| def validate_sensor_type(sensor_type: str) -> bool: | |
| """Validate sensor type against configured options""" | |
| is_valid, _, _ = get_validator().validate_sensor_type(sensor_type) | |
| return is_valid | |
| def validate_provincia(provincia: str) -> Tuple[bool, Optional[str]]: | |
| """Validate and normalize province parameter""" | |
| is_valid, normalized, _ = get_validator().validate_provincia(provincia) | |
| return is_valid, normalized | |
| def validate_zona(zona: str) -> bool: | |
| """Validate alert zone parameter""" | |
| is_valid, _, _ = get_validator().validate_zona(zona) | |
| return is_valid | |
| def validate_periodo(periodo: str) -> bool: | |
| """Validate time period parameter""" | |
| is_valid, _, _ = get_validator().validate_periodo(periodo) | |
| return is_valid | |
| # DEPRECATED: validate_mode_task_combination standalone function - mode parameter removed | |
| # def validate_mode_task_combination(mode: str, task: str) -> bool: | |
| # """Validate that mode and task combination is supported""" | |
| # is_valid, _ = get_validator().validate_mode_task_combination(mode, task) | |
| # return is_valid | |
| def get_valid_sensor_types() -> List[str]: | |
| """Get list of valid sensor types""" | |
| return get_validator().get_valid_sensor_types() | |
| def get_valid_provinces() -> Dict[str, str]: | |
| """Get mapping of province names to codes""" | |
| return get_validator().get_valid_provinces() | |
| def get_validation_errors(filters: Dict[str, Any]) -> List[str]: | |
| """ | |
| Validate a complete filter set and return any errors | |
| Args: | |
| filters: Dictionary of filters to validate | |
| Returns: | |
| List of validation error messages (empty if all valid) | |
| """ | |
| validator = get_validator() | |
| errors = [] | |
| validation_methods = { | |
| 'tipo_sensore': validator.validate_sensor_type, | |
| 'provincia': validator.validate_provincia, | |
| 'zona': validator.validate_zona, | |
| 'periodo': validator.validate_periodo, | |
| 'variabile_climatica': validator.validate_variabile_climatica | |
| } | |
| for filter_name, filter_value in filters.items(): | |
| if filter_name in validation_methods and filter_value: | |
| is_valid, _, suggestions = validation_methods[filter_name](filter_value) | |
| if not is_valid: | |
| error_msg = f"Invalid {filter_name}: '{filter_value}'" | |
| if suggestions: | |
| error_msg += f". Suggestions: {', '.join(suggestions[:3])}" | |
| errors.append(error_msg) | |
| return errors | |