From 08cfb39ad5322cf99086990a70a68a84826e8b30 Mon Sep 17 00:00:00 2001 From: Andrey Golikov Date: Wed, 22 Apr 2026 20:03:30 +0600 Subject: [PATCH] feat: complete task 6 --- task6/main.py | 200 +++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 182 insertions(+), 18 deletions(-) diff --git a/task6/main.py b/task6/main.py index 2677550..534b928 100644 --- a/task6/main.py +++ b/task6/main.py @@ -1,12 +1,69 @@ from collections import deque -from dataclasses import dataclass -from icecream import ic -from numpy import random -from typing import Final, Iterable +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 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 @@ -15,9 +72,65 @@ class Plane: Plane.__id_increment += 1 @dataclass -class Request: - time: float - plane: Plane +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") + +def reject_plane(plane: Plane) -> None: + stats.rejected_in_requests += 1 + print(f"Rejected incoming plane №{plane.id}") + +def launch_plane(plane: Plane, stats: Stats) -> None: + stats.launches_count += 1 + print(f"Launched plane №{plane.id}") + +def land_plane(plane: Plane, stats: Stats) -> None: + stats.landings_count += 1 + print(f"Plane №{plane.id} landed") def generate_timeline(incoming_planes_per_hour: float, outcoming_planes_per_hour: float, interval_minutes: int) -> Iterable[tuple[int, int]]: def _generate(planes_per_hour: float) -> NDArray: @@ -27,19 +140,70 @@ def generate_timeline(incoming_planes_per_hour: float, outcoming_planes_per_hour outcoming_planes_per_minute = _generate(outcoming_planes_per_hour) return zip(incoming_planes_per_minute, outcoming_planes_per_minute) -# takeoff_duration: Final[float] = 30 # minutes +def process_new_request(in_queue: Queue[Request], out_queue: Queue[Request], current_time: int, process_duration: int) -> ProcessedRequest: + plane: Optional[Plane] + if in_queue: + request = in_queue.pop() + plane = request.plane + stats.in_waiting_time.append(current_time - request.request_time) + print(f"{current_time}: Processing incoming plane №{plane.id}") + process_type = "landing" + else: + request = out_queue.pop() + plane = request.plane + stats.out_waiting_time.append(current_time - request.request_time) + print(f"{current_time}: Processing outcoming plane №{plane.id}") + process_type = "launching" + return ProcessedRequest(request.request_time, request.plane, current_time, current_time + process_duration, process_type) -# max_queue: Final[int] = int(input()) -# interval: Final[float] = int(input()) # minutes -# incoming_planes, outcoming_planes = map(float, input().split()) # per hour +takeoff_duration: Final[int] = int(input("Takeoff duration: ")) # minutes -# in_queue: deque[Plane] = deque(maxlen=max_queue) -# out_queue: deque[Plane] = deque(maxlen=max_queue) +max_queue: Final[int] = int(input("Max queue: ")) +interval: Final[int] = int(input("Interval: ")) # minutes +incoming_planes, outcoming_planes = map(float, input("Planes: ").split()) # per hour -timeline = generate_timeline(5,2, 120) +in_queue: Queue[Request] = EnforcingQueue[Request](max_queue) +out_queue: Queue[Request] = PermissiveQueue[Request](max_queue, warn_about_outcoming_queue_overload) + +current_request: Optional[ProcessedRequest] = None + +stats = Stats() +timeline = generate_timeline(5,2, interval) + +i = 0 for i, (in_planes_count, out_planes_count) in enumerate(timeline): - for a in range(in_planes_count): - ic(f"{i} in: {Plane().id}") - for a in range(out_planes_count): - ic(f"{i} out: {Plane().id}") + stats.total_plane_requests += in_planes_count + out_planes_count + stats.in_requests += in_planes_count + stats.out_requests += out_planes_count + + for _ in range(in_planes_count): + new_plane = Plane() + try: + in_queue.add(Request(i, new_plane)) + stats.accepted_in_requests += 1 + except IndexError: + reject_plane(new_plane) + stats.rejected_in_requests += 1 + for _ in range(out_planes_count): + new_plane = Plane() + out_queue.add(Request(i, new_plane)) + stats.accepted_out_requests += 1 + + if not current_request: + if len(in_queue) or len(out_queue): + current_request = process_new_request(in_queue, out_queue, i, takeoff_duration) + continue + stats.sleep_minutes += 1 + if current_request and current_request.process_time == i: + if current_request.type_ == "launching": + launch_plane(current_request.plane, stats) + else: + land_plane(current_request.plane, stats) + current_request = None + +stats.in_queued = len(in_queue) +stats.out_queued = len(out_queue) + +finalize_waiting_stats(stats, in_queue, out_queue, i) +show_stats(stats)