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 # Unique identifier for this run self.config = config # User-provided configuration self.status = SimulationStatus.IDLE # Initial state self.clock: float = 0.0 # Virtual simulation time # System state at current time instant self.tellers: Dict[str, Teller] = {} # All available tellers by ID self.waiting_queue: List[Customer] = [] # Customers waiting for service # Priority queue maintaining chronological timeline of future events self.event_queue: List[SimulationEvent] = [] # Component that generates arrival times and customer attributes _gen_config = {**self.config.arrival_config, **self.config.service_config} self.generator = ConfigurableGenerator(_gen_config)
simulation_id
Unique identifier (typically UUID) for this simulation instance
config
SimulationConfig object with all parameters (num_tellers, arrival_rate, etc.)
status
Current lifecycle state: IDLE, RUNNING, PAUSED, or FINISHED
clock
Virtual time counter (seconds). Jumps from event to event, NOT real-time!
tellers
Dictionary mapping teller IDs (“T-1”, “T-2”, …) to Teller entities
waiting_queue
List of Customer entities waiting for service, sorted by priority then arrival_time
event_queue
Min-heap of SimulationEvent objects ordered by time
generator
ConfigurableGenerator that produces arrival intervals, priorities, and service times
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))
Purpose: Reset simulation to initial state and seed the first event.
1
Reset state
Set status to IDLE, clock to 0.0
2
Clear queues
Empty waiting_queue and event_queue
3
Create tellers
Initialize num_tellers Teller entities with IDs T-1, T-2, …
4
Schedule first arrival
Generate random arrival time and add ARRIVAL event to queue
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) # Check termination if current_event.time > self.config.max_simulation_time: break # Advance clock self.clock = current_event.time self.process_next_event(current_event) self.status = SimulationStatus.FINISHED
Purpose: Main event processing loop.
1
Set status to RUNNING
Indicates simulation is active
2
Loop while events exist
Continue until event_queue is empty or status changes (e.g., PAUSED)
3
Pop earliest event
Extract minimum-time event from heap: heapq.heappop(event_queue)
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)
Purpose: Add future event to the timeline.
Time complexity: O(log n)
Maintains heap invariant: Smallest event.time at index 0
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)
def handle_service_start(self, teller_id: str, customer: Customer) -> None: """ Inicia oficialmente el tiempo de ventanilla entre un cajero y un cliente particular. Programa a futuro el evento que indicará el final del trámite. """ teller = self.tellers.get(teller_id) if teller: teller.start_service(customer, self.clock) # Schedule service completion end_time = self.clock + customer.service_time self.schedule_event(SimulationEvent(end_time, EventType.SERVICE_END, teller_id=teller_id))
def handle_service_end(self, teller_id: str) -> None: """ Registra el momento en el que el cliente actual se retira de la ventanilla, completando su transacción. A continuación, habilita inmediatamente al cajero para atender al siguiente en la cola (si hubiera). """ teller = self.tellers.get(teller_id) if teller: teller.end_service() # Teller now IDLE, try to assign next customer self._assign_free_teller()
Detailed Flow
1
Get teller entity
Lookup teller by ID
2
End service
customer.status = “COMPLETED”
teller.current_customer = None
teller.status = IDLE
teller.sessions_served += 1
3
Attempt next assignment
Call _assign_free_teller()
If someone waiting, immediately schedule their SERVICE_START
def _assign_free_teller(self) -> None: """ Busca secuencialmente si existe una ventanilla inactiva (IDLE). En caso de encontrar una y haber gente esperando, extrae al primer cliente de la fila y programa el evento SERVICE_START para esa ventanilla en el reloj actual. """ if not self.waiting_queue: return for t_id, teller in self.tellers.items(): if teller.status == "IDLE" or getattr(teller.status, "value", None) == "IDLE": next_customer = self.waiting_queue.pop(0) # Schedule immediate SERVICE_START self.schedule_event(SimulationEvent(self.clock, EventType.SERVICE_START, customer=next_customer, teller_id=t_id)) return # Assign only one customer per call
Purpose: Match waiting customers with idle tellers.
This method assigns only ONE customer per call. It must be called after EVERY event that could free a teller or add a customer.