From 17d27f0605349ba772c507e210e04f5c0908bec8 Mon Sep 17 00:00:00 2001 From: Andrey Golikov Date: Wed, 22 Apr 2026 20:13:35 +0600 Subject: [PATCH] refactor: move some classes and functions to separate files --- task6/entities.py | 21 ++++++++ task6/main.py | 121 +++------------------------------------------- task6/queues.py | 50 +++++++++++++++++++ task6/stats.py | 53 ++++++++++++++++++++ 4 files changed, 130 insertions(+), 115 deletions(-) create mode 100644 task6/entities.py create mode 100644 task6/queues.py create mode 100644 task6/stats.py diff --git a/task6/entities.py b/task6/entities.py new file mode 100644 index 0000000..2d82a91 --- /dev/null +++ b/task6/entities.py @@ -0,0 +1,21 @@ +from dataclasses import dataclass +from typing import Literal + + +@dataclass +class Request: + request_time: int + plane: Plane + +@dataclass +class ProcessedRequest(Request): + start_process_time: int + process_time: int + type_: Literal["landing", "launching"] + +class Plane: + __id_increment = 0 + + def __init__(self) -> None: + self.id = self.__id_increment + Plane.__id_increment += 1 diff --git a/task6/main.py b/task6/main.py index 534b928..8e5f7d0 100644 --- a/task6/main.py +++ b/task6/main.py @@ -1,122 +1,13 @@ -from collections import deque -from dataclasses import dataclass, field -from numpy import random, average -from typing import Callable, Final, Iterable, Literal, Optional, Self, override -from abc import ABC, abstractmethod +from stats import Stats, finalize_waiting_stats, show_stats +from entities import Plane, ProcessedRequest, Request +from queues import EnforcingQueue, PermissiveQueue, Queue + +from numpy import random +from typing import Final, Iterable, Optional from numpy.typing import NDArray -@dataclass -class Request: - request_time: int - plane: Plane - -@dataclass -class ProcessedRequest(Request): - start_process_time: int - process_time: int - type_: Literal["landing", "launching"] - -class Queue[T](ABC): - def __init__(self, size: int) -> None: - self.size = size - self.queue: deque[T] = deque() - - def __len__(self) -> int: - return len(self.queue) - - def __bool__(self) -> bool: - return bool(self.queue) - - @abstractmethod - def add(self, item: T) -> None: pass - - @abstractmethod - def pop(self) -> T: pass - -class PermissiveQueue[T](Queue): - def __init__(self, size: int, overload_function: Optional[Callable[[Self], None]] = None) -> None: - super().__init__(size) - self._on_overload = overload_function - - @override - def add(self, item: T) -> None: - self.queue.append(item) - if self._on_overload and len(self.queue) > self.size: - self._on_overload(self) - - @override - def pop(self) -> T: - return self.queue.popleft() - -class EnforcingQueue[T](Queue): - def __init__(self, size: int) -> None: - super().__init__(size) - - @override - def add(self, item: T) -> None: - if len(self.queue) >= self.size: - raise IndexError - self.queue.append(item) - - @override - def pop(self) -> T: - return self.queue.popleft() - -class Plane: - __id_increment = 0 - - def __init__(self) -> None: - self.id = self.__id_increment - Plane.__id_increment += 1 - -@dataclass -class Stats: - total_plane_requests = 0 - - in_requests = 0 - out_requests = 0 - - accepted_in_requests = 0 - accepted_out_requests = 0 - - rejected_in_requests = 0 - rejected_out_requests = 0 - - landings_count = 0 - launches_count = 0 - - in_queued = 0 - out_queued = 0 - - sleep_minutes = 0 - - in_waiting_time: list[float] = field(default_factory=list) - out_waiting_time: list[float] = field(default_factory=list) - -def show_stats(stats: Stats) -> None: - string = f"""============ -Total requests: {stats.total_plane_requests} -In requests: {stats.in_requests} -Out requests: {stats.out_requests} -Accepted in requests: {stats.accepted_in_requests} -Accepted out requests: {stats.accepted_out_requests} -Rejected in/out: {stats.rejected_in_requests}/{stats.rejected_out_requests} -Landings: {stats.landings_count} -Launches: {stats.launches_count} -Left in queues (in/out): {stats.in_queued}/{stats.out_queued} -Sleep: {stats.sleep_minutes} -Average waiting time (in/out): {average(stats.in_waiting_time)}/{average(stats.out_waiting_time)} -""" - print(string) - -def finalize_waiting_stats(stats: Stats, in_queue: Queue[Request], out_queue: Queue[Request], current_time: int) -> None: - for item in in_queue.queue: - stats.in_waiting_time.append(current_time - item.request_time) - for item in out_queue.queue: - stats.out_waiting_time.append(current_time - item.request_time) - def warn_about_outcoming_queue_overload(queue: Queue) -> None: print(f"Outcoming queue is overloaded: {len(queue)} planes/{queue.size} planes") diff --git a/task6/queues.py b/task6/queues.py new file mode 100644 index 0000000..93a2bdc --- /dev/null +++ b/task6/queues.py @@ -0,0 +1,50 @@ +from abc import ABC, abstractmethod +from collections import deque +from typing import Callable, Optional, Self, override + + +class Queue[T](ABC): + def __init__(self, size: int) -> None: + self.size = size + self.queue: deque[T] = deque() + + def __len__(self) -> int: + return len(self.queue) + + def __bool__(self) -> bool: + return bool(self.queue) + + @abstractmethod + def add(self, item: T) -> None: pass + + @abstractmethod + def pop(self) -> T: pass + +class PermissiveQueue[T](Queue): + def __init__(self, size: int, overload_function: Optional[Callable[[Self], None]] = None) -> None: + super().__init__(size) + self._on_overload = overload_function + + @override + def add(self, item: T) -> None: + self.queue.append(item) + if self._on_overload and len(self.queue) > self.size: + self._on_overload(self) + + @override + def pop(self) -> T: + return self.queue.popleft() + +class EnforcingQueue[T](Queue): + def __init__(self, size: int) -> None: + super().__init__(size) + + @override + def add(self, item: T) -> None: + if len(self.queue) >= self.size: + raise IndexError + self.queue.append(item) + + @override + def pop(self) -> T: + return self.queue.popleft() \ No newline at end of file diff --git a/task6/stats.py b/task6/stats.py new file mode 100644 index 0000000..38d96f1 --- /dev/null +++ b/task6/stats.py @@ -0,0 +1,53 @@ +from dataclasses import dataclass, field + +from numpy import average + +from entities import Request +from queues import Queue + + +@dataclass +class Stats: + total_plane_requests = 0 + + in_requests = 0 + out_requests = 0 + + accepted_in_requests = 0 + accepted_out_requests = 0 + + rejected_in_requests = 0 + rejected_out_requests = 0 + + landings_count = 0 + launches_count = 0 + + in_queued = 0 + out_queued = 0 + + sleep_minutes = 0 + + in_waiting_time: list[float] = field(default_factory=list) + out_waiting_time: list[float] = field(default_factory=list) + +def show_stats(stats: Stats) -> None: + string = f"""============ +Total requests: {stats.total_plane_requests} +In requests: {stats.in_requests} +Out requests: {stats.out_requests} +Accepted in requests: {stats.accepted_in_requests} +Accepted out requests: {stats.accepted_out_requests} +Rejected in/out: {stats.rejected_in_requests}/{stats.rejected_out_requests} +Landings: {stats.landings_count} +Launches: {stats.launches_count} +Left in queues (in/out): {stats.in_queued}/{stats.out_queued} +Sleep: {stats.sleep_minutes} +Average waiting time (in/out): {average(stats.in_waiting_time)}/{average(stats.out_waiting_time)} +""" + print(string) + +def finalize_waiting_stats(stats: Stats, in_queue: Queue[Request], out_queue: Queue[Request], current_time: int) -> None: + for item in in_queue.queue: + stats.in_waiting_time.append(current_time - item.request_time) + for item in out_queue.queue: + stats.out_waiting_time.append(current_time - item.request_time)