210 lines
6.7 KiB
Python
210 lines
6.7 KiB
Python
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 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")
|
|
|
|
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:
|
|
return random.poisson(planes_per_hour / 60, interval_minutes)
|
|
|
|
incoming_planes_per_minute = _generate(incoming_planes_per_hour)
|
|
outcoming_planes_per_minute = _generate(outcoming_planes_per_hour)
|
|
return zip(incoming_planes_per_minute, outcoming_planes_per_minute)
|
|
|
|
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)
|
|
|
|
takeoff_duration: Final[int] = int(input("Takeoff duration: ")) # minutes
|
|
|
|
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
|
|
|
|
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):
|
|
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)
|