Spaces:
Sleeping
Sleeping
File size: 6,400 Bytes
9e62f55 | 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 131 132 133 134 135 136 137 138 139 140 141 142 143 | 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() |