Spaces:
Sleeping
Sleeping
| import zlib | |
| import numpy as np | |
| from src.models.individual import ShiftPatterns | |
| from src.config import cfg | |
| from src.utils.hf_storage import load_json | |
| def load_employees_from_json(activity_folder): | |
| """ | |
| Hydration del modello di dominio anagrafico. | |
| Recupera i dati dal Dataset remoto e istanzia i fenotipi dei turni (maschere VDT). | |
| """ | |
| data = load_json(activity_folder, "employees.json") | |
| if not data: | |
| return [] | |
| employees = [] | |
| patterns = ShiftPatterns() | |
| for record in data: | |
| w_hours = float(record.get('work_hours', 8.0)) | |
| l_min = int(record.get('break_duration', 0)) | |
| # Hashing deterministico dell'ID per garantire che la stessa singola risorsa | |
| # mantenga sempre la stessa firma fenotipica (riproducibilità degli spezzati) | |
| seed = zlib.adler32(record['id'].encode('utf-8')) | |
| mask = patterns.get_mask_dynamic(w_hours, l_min, variant_seed=seed) | |
| # Parsing flessibile del target contrattuale (backward compatibility) | |
| mix = record.get('shift_mix', record.get('weekly_mix', {"WORK": 5, "OFF": 2})) | |
| target_days = int(mix["WORK"]) if "WORK" in mix else 7 - int(mix.get("OFF", 2)) | |
| employees.append({ | |
| "id": record['id'], | |
| "contract": record['contract'], | |
| "mask": mask, | |
| "shift_len": len(mask), | |
| "target_days": target_days, | |
| "constraints": record.get("constraints", {}) | |
| }) | |
| return employees | |
| def check_hours_balance(employees, target_matrix): | |
| """ | |
| Validazione strutturale pre-ottimizzazione. | |
| Calcola la capacità produttiva netta (escludendo shrinkage come pause VDT/Pranzo) | |
| e la confronta con il total workload richiesto dalla Demand. | |
| """ | |
| slot_hours = cfg.system_slot_minutes / 60.0 | |
| total_demand_slots = target_matrix.sum() | |
| total_demand_hours = total_demand_slots * slot_hours | |
| total_capacity_slots = 0 | |
| for emp in employees: | |
| # Calcolo della capacità netta computando solo i flag attivi (1) nella maschera booleana | |
| work_slots = np.sum(emp['mask']) | |
| days_per_week = emp.get('target_days', 5) | |
| total_capacity_slots += (work_slots * days_per_week) | |
| total_capacity_hours = total_capacity_slots * slot_hours | |
| print("\n[*] Analisi Strutturale: Bilancio Capacity vs Demand") | |
| print(f" ├─ Target Workload: {total_demand_hours:,.1f} h") | |
| print(f" ├─ Net Capacity: {total_capacity_hours:,.1f} h") | |
| diff = total_capacity_hours - total_demand_hours | |
| if diff >= 0: | |
| surplus_perc = (diff / total_demand_hours) * 100 | |
| print(f" └─ STATO: [OK] Surplus Teorico (+{diff:.1f}h / +{surplus_perc:.1f}%)") | |
| else: | |
| deficit_perc = (abs(diff) / total_demand_hours) * 100 | |
| print(f" └─ STATO: [WARN] Deficit Strutturale (-{abs(diff):.1f}h / -{deficit_perc:.1f}%)") | |
| return diff | |
| def minutes_to_time(slot_minutes_count): | |
| """ | |
| Utility per la conversione dello slot offset in timestamp leggibile (HH:MM). | |
| """ | |
| start_h = cfg.client_settings.get('day_start_hour', 8) | |
| total_minutes = (start_h * 60) + slot_minutes_count | |
| # Safe check per il wrap-around della mezzanotte (modulo 24h) per turni notturni | |
| total_minutes = total_minutes % (24 * 60) | |
| h = int(total_minutes // 60) | |
| m = int(total_minutes % 60) | |
| return f"{h:02d}:{m:02d}" |