Spaces:
Sleeping
Sleeping
| import numpy as np | |
| from src.config import cfg | |
| class ShiftPatterns: | |
| """ | |
| Modellazione del fenotipo dei turni. | |
| Mappa le regole contrattuali e i vincoli normativi (es. pause VDT) in tensori 1D | |
| che verranno successivamente proiettati sulla matrice di coverage. | |
| """ | |
| def __init__(self): | |
| # Parametri normativi (es. Pausa ogni 2 ore per i video-terminalisti) | |
| self.vdt_interval_min = cfg.system_settings.get('vdt_interval_minutes', 120) | |
| self.vdt_break_min = cfg.system_settings.get('vdt_break_minutes', 15) | |
| # Quantizzazione: trasformazione dai minuti agli slot di sistema | |
| self.max_work_slots = int(self.vdt_interval_min / cfg.system_slot_minutes) | |
| self.vdt_slots = max(1, int(self.vdt_break_min / cfg.system_slot_minutes)) | |
| def _create_block_inside_vdt(self, duration_minutes, flip_strategy=False): | |
| """ | |
| Genera un macro-blocco lavorativo iniettando le micro-pause VDT obbligatorie. | |
| """ | |
| total_slots = int(round(duration_minutes / cfg.system_slot_minutes)) | |
| mask = np.ones(total_slots, dtype=int) | |
| # Early exit: se il turno è più corto dell'intervallo VDT, niente pause | |
| if total_slots <= self.max_work_slots: | |
| return mask | |
| # Sliding window per "scavare" gli slot di pausa all'interno della maschera booleana | |
| cursor = 0 | |
| while cursor < total_slots: | |
| break_start_idx = cursor + self.max_work_slots | |
| if break_start_idx >= total_slots: | |
| break | |
| break_end_idx = min(break_start_idx + self.vdt_slots, total_slots) | |
| mask[break_start_idx : break_end_idx] = 0 | |
| cursor = break_end_idx | |
| # Il flip garantisce varianza fenotipica a parità di orario (es. pausa all'inizio vs alla fine) | |
| return np.flip(mask) if flip_strategy else mask | |
| def get_mask_dynamic(self, work_hours, lunch_minutes, variant_seed=0): | |
| """ | |
| Costruisce la firma oraria completa (fenotipo) del dipendente. | |
| Gestisce dinamicamente split-shift e variazioni deterministiche tramite seed. | |
| """ | |
| total_contract_min = work_hours * 60 | |
| lunch_slots = int(round(lunch_minutes / cfg.system_slot_minutes)) | |
| # Decodifica bit a bit del seed per alterare la struttura delle pause | |
| # mantenendo un output deterministico per lo stesso ID dipendente | |
| flip_p1 = (variant_seed % 2) != 0 | |
| flip_p2 = ((variant_seed >> 1) % 2) != 0 | |
| # Vincolo normativo: i turni lunghi richiedono uno spezzato (es. mattina / pomeriggio) | |
| if work_hours > 6: | |
| first_part_min = 4 * 60 # Blocco standard pre-pranzo | |
| second_part_min = total_contract_min - first_part_min | |
| has_lunch = (lunch_slots > 0) | |
| else: | |
| first_part_min = total_contract_min | |
| second_part_min = 0 | |
| has_lunch = False | |
| mask_part1 = self._create_block_inside_vdt(first_part_min, flip_strategy=flip_p1) | |
| mask_part2 = self._create_block_inside_vdt(second_part_min, flip_strategy=flip_p2) if second_part_min > 0 else np.array([], dtype=int) | |
| # Assemblaggio vettoriale del turno completo | |
| components = [mask_part1] | |
| if has_lunch: | |
| components.append(np.zeros(lunch_slots, dtype=int)) | |
| if len(mask_part2) > 0: | |
| components.append(mask_part2) | |
| return np.concatenate(components) |