Hugging Face's logo

Spaces:
NextGenTech
/
GeneticWFM
Sleeping

App Files Community
GeneticWFM
File size: 3,460 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
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}"