Spaces:
Sleeping
Sleeping
File size: 3,601 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 | 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 |