Spaces:
Sleeping
Sleeping
| import json | |
| import os | |
| from src.utils.hf_storage import load_json | |
| class ConfigManager: | |
| """ | |
| Singleton implementation for global configuration state management. | |
| Gestisce il caricamento a cascata (Cascade Loading) dei parametri di sistema e di dominio. | |
| """ | |
| _instance = None | |
| def __new__(cls, activity_name=None): | |
| if cls._instance is None: | |
| cls._instance = super(ConfigManager, cls).__new__(cls) | |
| cls._instance.initialized = False | |
| return cls._instance | |
| def __init__(self, activity_name=None): | |
| # Lazy Initialization: previene la sovrascrittura dello stato su chiamate multiple | |
| if getattr(self, 'initialized', False) and not activity_name: | |
| return | |
| # --- L0: HARDCODED FALLBACKS (Safety Net) --- | |
| self.activity_name = None | |
| self.system_slot_minutes = 15 | |
| self.planning_slot_minutes = 30 | |
| self.expansion_factor = 2 | |
| self.daily_slots = 96 | |
| # Safe allocations per le strutture di dominio | |
| self.client_settings = {"day_start_hour": 8, "day_end_hour": 20} | |
| self.system_settings = {"vdt_interval_minutes": 120, "vdt_break_minutes": 15} | |
| self.weights = {"understaffing": 1000, "overstaffing": 10, "homogeneity": 20, "soft_preference": 50} | |
| # Hyper-parametri di default per l'Engine Genetico | |
| self.genetic_params = { | |
| "population_size": 200, "generations": 100, "mutation_rate": 0.3, | |
| "crossover_rate": 0.8, "elitism_rate": 0.02, "tournament_size": 5, | |
| "heuristic_rate": 0.8, "guided_mutation_split": 0.4, "heuristic_noise": 0.2 | |
| } | |
| # Inizializzazione sicura del routing orario | |
| self.operating_hours = {"default": "09:00-18:00", "exceptions": {}} | |
| self.hours = self.operating_hours | |
| if activity_name: | |
| self.load_configurations(activity_name) | |
| else: | |
| print("[WARN] ConfigManager istanziato senza context. Approvvigionamento defaults (L0) completato.") | |
| self.initialized = True | |
| def load_configurations(self, activity_name): | |
| """Orchestratore del configuration loading a 3 livelli (L0 -> L1 -> L2).""" | |
| self.activity_name = activity_name | |
| base_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| # --- L1: ENGINE CONFIG (Global System Overrides) --- | |
| engine_path = os.path.join(base_path, "src", "config", "engine_config.json") | |
| try: | |
| if os.path.exists(engine_path): | |
| with open(engine_path, 'r') as f: | |
| engine_data = json.load(f) | |
| if 'system_settings' in engine_data: | |
| self.system_settings.update(engine_data['system_settings']) | |
| self.system_slot_minutes = self.system_settings.get('system_slot_minutes', 15) | |
| if 'genetic_params' in engine_data: | |
| self.genetic_params.update(engine_data['genetic_params']) | |
| except Exception as e: | |
| print(f"[WARN] Impossibile risolvere L1 engine_config.json ({e}). Proceeding with L0.") | |
| # --- L2: ACTIVITY CONFIG (Tenant/Domain Specifics via Object Storage) --- | |
| activity_data = load_json(activity_name, "activity_config.json") | |
| if not activity_data: | |
| print(f"[FATAL] Configurazione L2 mancante sul Dataset HF per l'attività '{activity_name}'.") | |
| return | |
| # Merging dei layer applicativi | |
| self.client_settings = activity_data.get('client_settings', self.client_settings) | |
| self.planning_slot_minutes = self.client_settings.get('planning_slot_minutes', 30) | |
| if 'weights' in activity_data: | |
| self.weights = activity_data['weights'] | |
| if 'operating_hours' in activity_data: | |
| self.operating_hours = activity_data['operating_hours'] | |
| self.hours = self.operating_hours | |
| if 'genetic_params' in activity_data: | |
| self.genetic_params.update(activity_data['genetic_params']) | |
| # --- DERIVED METRICS COMPUTATION --- | |
| self.slot_minutes = self.system_slot_minutes | |
| self.expansion_factor = int(self.planning_slot_minutes / self.system_slot_minutes) | |
| if self.expansion_factor < 1: | |
| self.expansion_factor = 1 | |
| # Calcolo dimensione tensore giornaliero | |
| start = self.client_settings.get('day_start_hour', 8) | |
| end = self.client_settings.get('day_end_hour', 20) | |
| self.daily_slots = int((end - start) * 60 / self.system_slot_minutes) | |
| if self.daily_slots <= 0: | |
| self.daily_slots = 96 | |
| self.initialized = True | |
| print(f"[OK] State Sync completato: {activity_name} (Shift Bounds: {start}:00-{end}:00, Grid: {self.daily_slots} slots)") | |
| def get_closing_slot(self, day_idx): | |
| """Mappa l'orario di chiusura algebrico sull'indice dello slot di sistema.""" | |
| day_str = str(day_idx) | |
| exceptions = self.hours.get('exceptions', {}) | |
| # 1. Rule Extraction (Exception override vs Baseline) | |
| if day_str in exceptions: | |
| time_range = exceptions[day_str] | |
| else: | |
| time_range = self.hours.get('default', "09:00-18:00") | |
| # 2. Explicit closure flag | |
| if str(time_range).strip().upper() == "CLOSED": | |
| return 0 | |
| try: | |
| # 3. Time-to-Slot Quantization | |
| _, close_time = time_range.split('-') | |
| h, m = map(int, close_time.split(':')) | |
| start_h = self.client_settings.get('day_start_hour', 8) | |
| minutes_from_start = (h - start_h) * 60 + m | |
| divisor = self.system_slot_minutes if self.system_slot_minutes > 0 else 15 | |
| slot_idx = int(minutes_from_start / divisor) | |
| return max(0, slot_idx) | |
| except Exception as e: | |
| # Failsafe: Parsing error mappato come hard closure per prevenire out-of-bounds nell'engine C/Numba | |
| print(f"[ERROR] Constraint parsing fallito sul Day {day_idx} ('{time_range}'): {e}. Forzatura status CLOSED.") | |
| return 0 | |
| def is_day_closed(self, day_idx): | |
| return self.get_closing_slot(day_idx) == 0 | |
| # Istanza globale esportata per i moduli downstream | |
| cfg = ConfigManager() |