Spaces:
Sleeping
Sleeping
| # ===================================================== | |
| # Apckeyl Framework | |
| # Version 1.1 | |
| # telegram_queue.py | |
| # ===================================================== | |
| """ | |
| Telegram Queue. | |
| Соединяет Telegram Framework | |
| с системой очереди проекта. | |
| Version 1.1 | |
| """ | |
| from priority_queue import PriorityQueue | |
| from batch_processor import BatchProcessor | |
| class TelegramQueue: | |
| """ | |
| Telegram Queue. | |
| Принимает задания Telegram, | |
| помещает их в PriorityQueue | |
| и может передавать их | |
| BatchProcessor. | |
| Сам AI не выполняет. | |
| """ | |
| def __init__(self): | |
| self.queue = PriorityQueue() | |
| self.batch_processor = BatchProcessor() | |
| # ------------------------------------------------- | |
| def add_task( | |
| self, | |
| task, | |
| owner=False, | |
| ): | |
| self.queue.add( | |
| task=task, | |
| owner=owner, | |
| ) | |
| print() | |
| print("Task added to PriorityQueue.") | |
| print( | |
| f"Queue size : {self.queue.size()}" | |
| ) | |
| # ------------------------------------------------- | |
| def process_next( | |
| self, | |
| mode, | |
| quality, | |
| denoise, | |
| ): | |
| task = self.queue.get_next() | |
| if task is None: | |
| print() | |
| print("Queue is empty.") | |
| return None | |
| print() | |
| print("Processing task...") | |
| return self.batch_processor.process_with_limit( | |
| images=[task], | |
| mode=mode, | |
| quality=quality, | |
| denoise=denoise, | |
| ) | |
| # ------------------------------------------------- | |
| def size(self): | |
| return self.queue.size() | |
| # ------------------------------------------------- | |
| def clear(self): | |
| self.queue.clear() | |
| print() | |
| print("Telegram queue cleared.") |