feat: add task 8 with inverted index

task6/entities.py

@@ -1,21 +0,0 @@
-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

task6/main.py

@@ -1,100 +0,0 @@
-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
-
-
-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)

task6/queues.py

@@ -1,50 +0,0 @@
-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()

task6/stats.py

@@ -1,53 +0,0 @@
-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)