import pygame import sys import random # Инициализация Pygame pygame.init() WIDTH, HEIGHT = 1024, 768 screen = pygame.display.set_mode((WIDTH, HEIGHT)) pygame.display.set_caption("Диспетчер потоков – управление режимами") clock = pygame.time.Clock() # Шрифты FONT_TITLE = pygame.font.Font(None, 48) FONT = pygame.font.Font(None, 36) FONT_SMALL = pygame.font.Font(None, 28) # Цвета BG = (20, 25, 45) QUEUE_BG = (35, 45, 65) TEXT_COL = (255, 255, 255) BUTTON_COL = (70, 130, 200) BUTTON_HOVER = (100, 160, 240) ACTIVE_COL = (255, 80, 80) MODAL_BG = (0, 0, 0, 200) # ---------- Класс потока ---------- class Thread: def __init__(self, tid, priority, burst): self.tid = tid self.priority = priority self.remaining = burst self.base_color = (random.randint(100, 220), random.randint(100, 220), 255) self.animation_progress = 0.0 # 0..1 для анимации выезда def get_color(self, is_active=False, pulse=0): if is_active: # Пульсирующий красный r = ACTIVE_COL[0] + int(pulse * 50) g = ACTIVE_COL[1] - int(pulse * 30) b = ACTIVE_COL[2] - int(pulse * 30) return (min(255, r), max(50, g), max(50, b)) return self.base_color # ---------- Планировщик ---------- class Scheduler: def __init__(self, algorithm): self.algorithm = algorithm # "rr" или "priority" self.queue = [] self.current = None self.time_quantum = 3 self.quantum_counter = 0 self.interrupts_handled = 0 self.total_steps = 0 def add_thread(self, thread): thread.animation_progress = 0.0 self.queue.append(thread) def step(self): self.total_steps += 1 if self.current is None: if not self.queue: return False self.current = self.queue.pop(0) self.quantum_counter = 0 return True self.current.remaining -= 1 self.quantum_counter += 1 if self.current.remaining <= 0: self.current = None self.quantum_counter = 0 elif self.algorithm == "rr" and self.quantum_counter >= self.time_quantum: self.queue.append(self.current) self.current = None self.quantum_counter = 0 elif self.algorithm == "priority": higher = [t for t in self.queue if t.priority < self.current.priority] if higher: self.queue.append(self.current) self.current = None self.quantum_counter = 0 return True def all_finished(self): return self.current is None and len(self.queue) == 0 # ---------- Вспомогательные функции отрисовки ---------- def draw_button(text, x, y, w, h, color=None, hover_color=None): if color is None: color = BUTTON_COL if hover_color is None: hover_color = BUTTON_HOVER mouse = pygame.mouse.get_pos() click = pygame.mouse.get_pressed() is_hover = (x < mouse[0] < x + w and y < mouse[1] < y + h) col = hover_color if is_hover else color pygame.draw.rect(screen, col, (x, y, w, h), border_radius=12) pygame.draw.rect(screen, (0, 0, 0), (x + 2, y + 2, w, h), border_radius=12, width=1) text_surf = FONT_SMALL.render(text, True, TEXT_COL) screen.blit(text_surf, (x + w // 2 - text_surf.get_width() // 2, y + h // 2 - text_surf.get_height() // 2)) return is_hover and click[0] def draw_queue_animated(threads, current, pulse): start_x = 80 start_y = 180 slot_w = 140 slot_h = 80 max_visible = 6 pygame.draw.rect(screen, QUEUE_BG, (50, start_y - 20, WIDTH - 100, 130), border_radius=15) for i, thr in enumerate(threads[:max_visible]): # Анимация выезда if thr.animation_progress < 1.0: thr.animation_progress = min(1.0, thr.animation_progress + 0.1) offset = int((1 - thr.animation_progress) * slot_w) x = start_x + i * slot_w - offset y = start_y rect = pygame.Rect(x, y, slot_w - 10, slot_h) color = thr.get_color(False) pygame.draw.rect(screen, color, rect, border_radius=10) pygame.draw.rect(screen, (255, 255, 255), rect, width=2, border_radius=10) text = FONT_SMALL.render(f"T{thr.tid} P:{thr.priority} {thr.remaining}", True, TEXT_COL) screen.blit(text, (rect.x + 5, rect.y + 30)) if current: # Пульсация активного потока scale = 1 + pulse * 0.05 w = int(160 * scale) h = int(80 * scale) x = WIDTH // 2 - w // 2 y = start_y - 70 - int(pulse * 5) color = current.get_color(True, pulse) rect = pygame.Rect(x, y, w, h) pygame.draw.rect(screen, color, rect, border_radius=15) pygame.draw.rect(screen, (255, 255, 200), rect, width=3, border_radius=15) text = FONT.render(f"Активен: T{current.tid} (ост.{current.remaining})", True, TEXT_COL) screen.blit(text, (rect.x + rect.w // 2 - text.get_width() // 2, rect.y + rect.h // 2 - text.get_height() // 2)) def modal_interrupt(scheduler): """Модальное окно выбора обработки прерывания""" modal_w, modal_h = 450, 220 modal_x = WIDTH // 2 - modal_w // 2 modal_y = HEIGHT // 2 - modal_h // 2 overlay = pygame.Surface((WIDTH, HEIGHT), pygame.SRCALPHA) overlay.fill((0, 0, 0, 180)) screen.blit(overlay, (0, 0)) pygame.draw.rect(screen, (50, 55, 80), (modal_x, modal_y, modal_w, modal_h), border_radius=20) pygame.draw.rect(screen, (100, 130, 200), (modal_x, modal_y, modal_w, modal_h), width=3, border_radius=20) title = FONT.render("⚠ Внешнее прерывание!", True, (255, 220, 100)) screen.blit(title, (modal_x + modal_w // 2 - title.get_width() // 2, modal_y + 30)) sub = FONT_SMALL.render("Переход в режим ядра. Что делать?", True, TEXT_COL) screen.blit(sub, (modal_x + modal_w // 2 - sub.get_width() // 2, modal_y + 70)) btn_imm = draw_button("Обработать немедленно", modal_x + 30, modal_y + 140, 180, 45) btn_delay = draw_button("Отложить", modal_x + 240, modal_y + 140, 180, 45) pygame.display.flip() while True: for event in pygame.event.get(): if event.type == pygame.QUIT: pygame.quit() sys.exit() if event.type == pygame.MOUSEBUTTONDOWN: if btn_imm: return True if btn_delay: return False # Обновляем состояние кнопок btn_imm = draw_button("Обработать немедленно", modal_x + 30, modal_y + 140, 180, 45) btn_delay = draw_button("Отложить", modal_x + 240, modal_y + 140, 180, 45) pygame.display.flip() clock.tick(30) # ---------- Основная функция ---------- def main(): scene = 0 # 0-выбор алгоритма, 1-добавление потоков, 2-диспетчеризация, 3-прерывание (модальное), 4-статистика algorithm = None scheduler = None next_tid = 1 message = "" step_count = 0 interrupt_available = False threads_added = [] running = True pulse = 0.0 pulse_dir = 1 while running: # Пульсация для активного потока pulse += 0.05 * pulse_dir if pulse >= 1.0: pulse = 1.0 pulse_dir = -1 elif pulse <= 0.0: pulse = 0.0 pulse_dir = 1 screen.fill(BG) # Заголовок title = FONT_TITLE.render("⚙ Управление режимом потоков", True, TEXT_COL) screen.blit(title, (WIDTH // 2 - title.get_width() // 2, 20)) # ------------------ СЦЕНА 0: ВЫБОР АЛГОРИТМА ------------------ if scene == 0: prompt = FONT.render("Выберите алгоритм планирования:", True, TEXT_COL) screen.blit(prompt, (WIDTH // 2 - prompt.get_width() // 2, 180)) if draw_button("Round Robin", WIDTH // 2 - 180, 280, 160, 60): algorithm = "rr" scheduler = Scheduler("rr") scene = 1 threads_added = [] next_tid = 1 if draw_button("Приоритетное", WIDTH // 2 + 20, 280, 160, 60): algorithm = "priority" scheduler = Scheduler("priority") scene = 1 threads_added = [] next_tid = 1 if draw_button("Выход", WIDTH - 120, HEIGHT - 70, 100, 50): running = False # ------------------ СЦЕНА 1: ДОБАВЛЕНИЕ ПОТОКОВ ------------------ elif scene == 1: info = FONT_SMALL.render("Добавьте минимум 3 потока (случайные параметры)", True, TEXT_COL) screen.blit(info, (WIDTH // 2 - info.get_width() // 2, 120)) if draw_button("+ Добавить поток", WIDTH - 200, HEIGHT - 100, 180, 50): new = Thread(next_tid, random.randint(1, 5), random.randint(4, 12)) scheduler.add_thread(new) threads_added.append(new) next_tid += 1 message = f"✅ Поток T{new.tid} (приор.{new.priority}, burst {new.remaining}) добавлен" # Список добавленных потоков y = 200 for thr in threads_added: txt = FONT_SMALL.render(f"T{thr.tid} | Приоритет: {thr.priority} | Остаток: {thr.remaining}", True, TEXT_COL) screen.blit(txt, (60, y)) pygame.draw.circle(screen, thr.base_color, (40, y + 12), 10) y += 45 if len(threads_added) >= 3: if draw_button("▶ Запустить планирование", WIDTH // 2 - 120, HEIGHT - 100, 240, 50): scene = 2 step_count = 0 interrupt_available = False # Сообщение и кнопка назад msg_surf = FONT_SMALL.render(message, True, (200, 220, 100)) screen.blit(msg_surf, (50, HEIGHT - 60)) if draw_button("Назад", 50, HEIGHT - 60, 100, 45): scene = 0 # ------------------ СЦЕНА 2: ДИСПЕТЧЕРИЗАЦИЯ ------------------ elif scene == 2: draw_queue_animated(scheduler.queue, scheduler.current, pulse) if draw_button("⏩ Шаг (квант)", WIDTH - 200, HEIGHT - 100, 170, 55): scheduler.step() step_count += 1 if scheduler.all_finished(): scene = 4 if step_count >= 4 and not interrupt_available: if draw_button("⚠ ПРЕРЫВАНИЕ", WIDTH - 200, HEIGHT - 180, 170, 50, (180, 60, 60), (220, 80, 80)): scene = 3 info_text = FONT_SMALL.render(f"Шагов: {step_count} | Алгоритм: {'RR' if algorithm == 'rr' else 'Priority'}", True, TEXT_COL) screen.blit(info_text, (30, HEIGHT - 50)) if draw_button("Выход", WIDTH - 100, 20, 80, 40): running = False # ------------------ СЦЕНА 3: ОБРАБОТКА ПРЕРЫВАНИЯ (МОДАЛЬНОЕ ОКНО) ------------------ elif scene == 3: immediate = modal_interrupt(scheduler) if immediate: if scheduler.current: old = scheduler.current.priority scheduler.current.priority = max(1, scheduler.current.priority - 2) scheduler.interrupts_handled += 1 message = f"✅ Прерывание обработано! Приоритет T{scheduler.current.tid} повышен с {old} до {scheduler.current.priority}" else: message = "Прерывание, но нет активного потока" else: message = "⏸ Прерывание отложено. Эффективность снижена." interrupt_available = True scene = 2 # ------------------ СЦЕНА 4: ФИНАЛЬНАЯ СТАТИСТИКА ------------------ elif scene == 4: if scheduler.interrupts_handled > 0: grade = "A (Отлично)" grade_color = (100, 255, 100) elif step_count < 10: grade = "B (Хорошо)" grade_color = (200, 255, 100) elif step_count < 20: grade = "C (Удовлетворительно)" grade_color = (255, 200, 100) else: grade = "D (Низкая эффективность)" grade_color = (255, 120, 120) stats = [ "📊 Статистика игры", f"Алгоритм: {'Round Robin' if algorithm == 'rr' else 'Приоритетное (вытеснение)'}", f"Всего шагов (квантов): {step_count}", f"Прерываний обработано немедленно: {scheduler.interrupts_handled}", f"Потоков завершено: {next_tid - 1}", f"Оценка: {grade}" ] y = 180 for line in stats: if "Оценка:" in line: txt = FONT.render(line, True, grade_color) else: txt = FONT.render(line, True, TEXT_COL) screen.blit(txt, (WIDTH // 2 - txt.get_width() // 2, y)) y += 55 if draw_button("🔄 Сыграть снова", WIDTH // 2 - 180, HEIGHT - 140, 160, 55): scene = 0 algorithm = None scheduler = None threads_added = [] next_tid = 1 step_count = 0 interrupt_available = False if draw_button("❌ Выход", WIDTH // 2 + 20, HEIGHT - 140, 160, 55): running = False pygame.display.flip() clock.tick(30) for event in pygame.event.get(): if event.type == pygame.QUIT: running = False pygame.quit() sys.exit() if __name__ == "__main__": main()