Spaces:
Sleeping
Sleeping
File size: 2,456 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 | import numpy as np
def align_demand_to_grid(raw_demand_row, current_config):
"""
Allinea la time-series della demand giornaliera alla griglia oraria di configurazione.
Gestisce dinamicamente truncation e padding in base a shift operativi di start/end.
"""
# Estrazione parametri di quantizzazione temporale
start_h = current_config['client_settings']['day_start_hour']
end_h = current_config['client_settings']['day_end_hour']
slot_min = current_config['client_settings']['planning_slot_minutes']
slots_per_day = int((end_h - start_h) * 60 / slot_min)
# Fallback strutturale per payload mancanti o corrotti
if not raw_demand_row:
return np.ones(slots_per_day, dtype=int)
# Early exit se la dimensionalità è già coerente con il setup
if len(raw_demand_row) == slots_per_day:
return np.array(raw_demand_row, dtype=int)
# --- RESHAPING PIPELINE ---
# Inizializzazione target array con baseline di staff (safety net = 1)
adjusted_demand = np.ones(slots_per_day, dtype=int)
# Euristica "Best Effort" per il mapping dei dati storici sulla nuova griglia.
# Assunto: invarianza del delta temporale (slot_min).
# TODO: In caso di mutazione della granularità (es. 15m -> 30m) serve un resampler/interpolatore.
limit = min(len(raw_demand_row), slots_per_day)
adjusted_demand[:limit] = raw_demand_row[:limit]
return adjusted_demand
def sanitize_weekly_demand(weekly_demand, current_config):
"""
Pipeline di sanitizzazione massiva per l'intera matrice settimanale.
Esegue stripping delle label testuali ed enforcing della consistenza dimensionale (7xN).
"""
sanitized = []
# Enforcing del vincolo a 7 giorni operativi
for i in range(7):
if i < len(weekly_demand):
row_data = weekly_demand[i]
# Stripping della label descrittiva se presente (es. "Giorno_0")
if isinstance(row_data[0], str):
row_vals = row_data[1:]
else:
row_vals = row_data
aligned = align_demand_to_grid(row_vals, current_config)
sanitized.append(aligned)
else:
# Imputazione di una curva flat di fallback per le giornate out-of-bounds
dummy = align_demand_to_grid([], current_config)
sanitized.append(dummy)
return sanitized |