DiscreteEventSimulation is the core paradigm powering SimulationBank. Unlike continuous simulation where time flows smoothly, DES advances time in discrete jumps from one event to the next.
Defined in backend/src/simulation/domain/simulation.py:14:
class DiscreteEventSimulation: """ Entidad raíz: representa una instancia completa de la simulación del banco. Actúa como el motor central que coordina el tiempo, procesa la cola de eventos y asigna los recursos. """ def __init__(self, simulation_id: str, config: SimulationConfig): self.simulation_id = simulation_id self.config = config self.status = SimulationStatus.IDLE self.clock: float = 0.0 # Virtual simulation clock # System state self.tellers: Dict[str, Teller] = {} self.waiting_queue: List[Customer] = [] # Event queue (priority queue ordered by time) self.event_queue: List[SimulationEvent] = [] # Customer generator using Poisson process self.generator = ConfigurableGenerator(_gen_config)
def initialize(self) -> None: """ Prepara el sistema antes de iniciar. Inicializa las ventanillas, vacía las colas y programa mecánicamente el primer cliente en llegar. """ self.status = SimulationStatus.IDLE self.clock = 0.0 self.waiting_queue.clear() self.event_queue.clear() # Initialize tellers for i in range(self.config.num_tellers): t_id = f"T-{i+1}" self.tellers[t_id] = Teller(id=t_id) # Schedule first arrival first_arrival_time = self.generator.get_next_arrival_interval() if first_arrival_time <= self.config.max_simulation_time: self.schedule_event(SimulationEvent(first_arrival_time, EventType.ARRIVAL, customer=None))
def run(self) -> None: """ Bucle principal del motor de simulación. Extrae cronológicamente el próximo evento, adelanta el reloj y ejecuta su lógica hasta agotar la cola de eventos o superar el tiempo máximo. """ self.status = SimulationStatus.RUNNING while self.event_queue and self.status == SimulationStatus.RUNNING: current_event = heapq.heappop(self.event_queue) # Get earliest event # Check termination if current_event.time > self.config.max_simulation_time: break # Advance clock to event time self.clock = current_event.time self.process_next_event(current_event) self.status = SimulationStatus.FINISHED
1
Extract next event
Pop the earliest event from the priority queue using heapq.heappop()
2
Advance clock
Jump virtual time to the event’s timestamp: self.clock = current_event.time
3
Process event
Execute event-specific logic that updates system state
4
Repeat
Loop until event queue is empty or max simulation time exceeded
def schedule_event(self, event: SimulationEvent) -> None: """ Añade un nuevo evento futuro a la línea de tiempo. Utiliza heappush para mantener la consistencia de eventos ordenados por tiempo. """ heapq.heappush(self.event_queue, event)
def process_next_event(self, event: SimulationEvent) -> None: """ Conmutador (switch) central que redirige el flujo de procesamiento dependiendo de si el evento es una llegada, un inicio de atención, o un fin de servicio. """ if event.event_type == EventType.ARRIVAL: self.handle_arrival() elif event.event_type == EventType.SERVICE_START: self.handle_service_start(event.teller_id, event.customer) elif event.event_type == EventType.SERVICE_END: self.handle_service_end(event.teller_id)