Spaces:
Sleeping
Sleeping
| import numpy as np | |
| from src.config import cfg | |
| # Mapping dei constraint per la vettorializzazione (flag numerici per il motore JIT) | |
| CONS_TYPE_NONE = 0 | |
| CONS_TYPE_HARD = 1 | |
| CONS_TYPE_SOFT = 2 | |
| CONS_TYPE_ABSENCE = 3 | |
| def process_demand(raw_data): | |
| """ | |
| Upsampling della demand operativa. | |
| Espande la granularità di business (es. slot da 30/60 min) | |
| sul clock interno di sistema (es. 15 min) per il calcolo matriciale. | |
| """ | |
| weekly_demand = np.zeros((7, cfg.daily_slots), dtype=int) | |
| ratio = int(cfg.expansion_factor) | |
| for day_idx, day_data in enumerate(raw_data): | |
| # Cast robusto dei valori input (fallback a 0 per dati sporchi/missing) | |
| input_vals = [int(x) if str(x).isdigit() else 0 for x in day_data[1:]] | |
| curr_sys_slot = 0 | |
| for val in input_vals: | |
| end_sys_slot = curr_sys_slot + ratio | |
| # Boundary check per evitare out-of-bounds se l'orario di chiusura varia | |
| if end_sys_slot <= cfg.daily_slots: | |
| weekly_demand[day_idx, curr_sys_slot:end_sys_slot] = val | |
| curr_sys_slot += ratio | |
| return weekly_demand | |
| def convert_employees_to_numpy(employees): | |
| """ | |
| Tensorizzazione dell'anagrafica. | |
| Estrae le liste di dizionari e crea array contigui in memoria (ndarrays) | |
| per bypassare l'overhead degli oggetti Python dentro i loop di Numba. | |
| """ | |
| num_emps = len(employees) | |
| # Fallback di sicurezza se l'anagrafica è vuota | |
| if num_emps == 0: | |
| return None, None, None, None, None | |
| # Ricerca dinamica della shift mask più lunga per dimensionare il tensore 2D | |
| max_len = max(len(e['mask']) for e in employees) | |
| # Inizializzazione tensori pre-allocati | |
| masks_matrix = np.zeros((num_emps, max_len), dtype=int) | |
| lengths_array = np.zeros(num_emps, dtype=int) | |
| target_days_array = np.zeros(num_emps, dtype=int) | |
| cons_type_matrix = np.zeros((num_emps, 7), dtype=int) | |
| cons_val_matrix = np.full((num_emps, 7), -1, dtype=int) | |
| def _to_slot(t_str): | |
| """Quantizzazione del timestamp testuale in indice slot di sistema.""" | |
| try: | |
| h, m = map(int, t_str.split(':')) | |
| minutes = (h - cfg.client_settings['day_start_hour']) * 60 + m | |
| return int(minutes / cfg.system_slot_minutes) | |
| except: | |
| return -1 # Fallback error code | |
| # Compilazione massiva delle matrici | |
| for i, emp in enumerate(employees): | |
| curr_len = len(emp['mask']) | |
| # Inserimento maschera con zero-padding implicito a destra | |
| masks_matrix[i, :curr_len] = emp['mask'] | |
| lengths_array[i] = curr_len | |
| # Recupero target contrattuale | |
| target_days_array[i] = emp.get('target_days', 5) | |
| # Mapping spaziale dei constraint (Hard/Soft/Assenze) | |
| constraints = emp.get('constraints', {}) | |
| for day_str, rule in constraints.items(): | |
| d = int(day_str) | |
| if d < 0 or d > 6: | |
| continue | |
| rtype = rule.get('type') | |
| if rtype == 'hard': | |
| cons_type_matrix[i, d] = CONS_TYPE_HARD | |
| cons_val_matrix[i, d] = _to_slot(rule.get('start_time', '00:00')) | |
| elif rtype == 'soft': | |
| cons_type_matrix[i, d] = CONS_TYPE_SOFT | |
| cons_val_matrix[i, d] = _to_slot(rule.get('start_time', '00:00')) | |
| elif rtype == 'absence': | |
| cons_type_matrix[i, d] = CONS_TYPE_ABSENCE | |
| return masks_matrix, lengths_array, target_days_array, cons_type_matrix, cons_val_matrix |