Parking Lot & Elevator System

import itertools
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional

# ---------- enums ----------
class VehicleType(Enum):
    BIKE = 1
    CAR = 2
    TRUCK = 3

class SpotType(Enum):          # SMALL fits a bike, MEDIUM a car, LARGE a truck
    SMALL = 1
    MEDIUM = 2
    LARGE = 3

FITS = {VehicleType.BIKE: SpotType.SMALL,
        VehicleType.CAR: SpotType.MEDIUM,
        VehicleType.TRUCK: SpotType.LARGE}

@dataclass
class Vehicle:
    number: str
    type: VehicleType

# ---------- spot & floor ----------
class ParkingSpot:
    def __init__(self, spot_id: str, spot_type: SpotType):
        self.id, self.type = spot_id, spot_type
        self.vehicle: Optional[Vehicle] = None   # the real car sits here, not just its id

    def is_free(self) -> bool:
        return self.vehicle is None

    def can_fit(self, v: Vehicle) -> bool:
        return self.is_free() and FITS[v.type] == self.type

    def park(self, v: Vehicle) -> None:
        self.vehicle = v

    def unpark(self) -> None:
        self.vehicle = None

class ParkingFloor:
    def __init__(self, floor_id: int, spots: list[ParkingSpot]):
        self.id, self.spots = floor_id, spots

    def find_spot(self, v: Vehicle) -> Optional[ParkingSpot]:
        for spot in self.spots:            # simple scan is fine in an interview
            if spot.can_fit(v):
                return spot
        return None

# ---------- ticket ----------
@dataclass
class Ticket:
    id: int
    vehicle: Vehicle
    spot: ParkingSpot
    entry_time: datetime = field(default_factory=datetime.now)
    exit_time: Optional[datetime] = None

    def mark_exit(self, when: Optional[datetime] = None) -> None:
        self.exit_time = when or datetime.now()

    def duration(self) -> timedelta:
        return (self.exit_time or datetime.now()) - self.entry_time

# ---------- strategies ----------
class FeeStrategy(ABC):
    @abstractmethod
    def calculate(self, ticket: Ticket) -> float: ...

class HourlyFee(FeeStrategy):
    RATE = {VehicleType.BIKE: 10, VehicleType.CAR: 20, VehicleType.TRUCK: 50}
    def calculate(self, ticket: Ticket) -> float:
        hours = max(1, -(-int(ticket.duration().total_seconds()) // 3600))  # round up, min 1h
        return hours * self.RATE[ticket.vehicle.type]

class FixedFee(FeeStrategy):
    def calculate(self, ticket: Ticket) -> float:
        return 50.0

class PaymentStrategy(ABC):
    @abstractmethod
    def pay(self, amount: float) -> None: ...

class UpiPayment(PaymentStrategy):
    def pay(self, amount: float) -> None:
        print(f"Paid Rs {amount:.0f} via UPI")

# ---------- the orchestrator ----------
class ParkingLot:
    def __init__(self):
        self.floors: list[ParkingFloor] = []
        self.active_tickets: dict[str, Ticket] = {}   # vehicle number -> ticket
        self._ids = itertools.count(1)
        self._lock = threading.Lock()                 # two gates must never get the same spot

    def add_floor(self, floor: ParkingFloor) -> None:
        self.floors.append(floor)

    def assign_spot(self, v: Vehicle) -> Optional[Ticket]:
        with self._lock:                              # find + park is ONE atomic step
            for floor in self.floors:
                spot = floor.find_spot(v)
                if spot:
                    spot.park(v)
                    ticket = Ticket(next(self._ids), v, spot)
                    self.active_tickets[v.number] = ticket
                    return ticket
            return None

    def release(self, vehicle_number: str) -> Ticket:
        with self._lock:
            ticket = self.active_tickets.pop(vehicle_number)
            ticket.spot.unpark()
            return ticket

# ---------- gates: thin API layers over the lot ----------
class EntryGate:
    def __init__(self, lot: ParkingLot):
        self.lot = lot

    def enter(self, v: Vehicle) -> Optional[Ticket]:
        ticket = self.lot.assign_spot(v)
        print(f"Ticket {ticket.id}: {v.number} -> spot {ticket.spot.id}" if ticket else "Lot full")
        return ticket

class ExitGate:
    def __init__(self, lot: ParkingLot, fee: FeeStrategy, payment: PaymentStrategy):
        self.lot, self.fee, self.payment = lot, fee, payment

    def exit(self, ticket: Ticket) -> float:
        ticket.mark_exit()
        amount = self.fee.calculate(ticket)
        self.payment.pay(amount)
        self.lot.release(ticket.vehicle.number)       # vacate the slot + drop active ticket
        return amount

# usage
lot = ParkingLot()
lot.add_floor(ParkingFloor(1, [ParkingSpot("1-S1", SpotType.SMALL),
                               ParkingSpot("1-M1", SpotType.MEDIUM)]))
entry = EntryGate(lot)
exit_gate = ExitGate(lot, HourlyFee(), UpiPayment())
t = entry.enter(Vehicle("KA-01-1234", VehicleType.CAR))   # Ticket 1: KA-01-1234 -> spot 1-M1
entry.enter(Vehicle("KA-02-9999", VehicleType.CAR))       # Lot full (only one medium spot)
exit_gate.exit(t)                                         # Paid Rs 20 via UPI

import heapq
import threading
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum

TICK = 0.05   # seconds to travel one floor (fast, for the demo)

class Direction(Enum):
    UP = "UP"
    DOWN = "DOWN"
    IDLE = "IDLE"

@dataclass(frozen=True)
class Request:
    floor_id: int     # "open the door at this floor" - hall button OR cabin button

class Elevator:
    """One lift = one thread. It knows nothing about the other lifts."""

    def __init__(self, elevator_id: int):
        self.id = elevator_id
        self.current_floor = 0
        self.direction = Direction.IDLE
        self.up_stops: list[int] = []      # min-heap: going up, nearest higher floor first
        self.down_stops: list[int] = []    # max-heap (store -floor): going down, nearest lower first
        self._cv = threading.Condition()   # a lock + wait/notify in one object
        self._running = True

    # ----- called by other threads (the system / passengers) -----
    def add_request(self, req: Request) -> None:
        with self._cv:
            if req.floor_id > self.current_floor:
                heapq.heappush(self.up_stops, req.floor_id)
            elif req.floor_id < self.current_floor:
                heapq.heappush(self.down_stops, -req.floor_id)
            else:
                print(f"Elevator {self.id} opens doors at {self.current_floor}")
            self._cv.notify()              # wake the lift if it was sleeping (idle)

    def stop(self) -> None:
        with self._cv:
            self._running = False
            self._cv.notify()

    # ----- the lift's own thread -----
    def run(self) -> None:
        while True:
            with self._cv:
                while self._running and not self.up_stops and not self.down_stops:
                    self.direction = Direction.IDLE
                    self._cv.wait()        # sleep instead of busy-spinning
                if not self._running:
                    return
                self._step()
            time.sleep(TICK)               # travel time, outside the lock

    def _step(self) -> None:
        # SCAN: keep the current direction while it has work, else switch.
        if self.direction != Direction.DOWN and self.up_stops:
            self.direction = Direction.UP
        elif self.down_stops:
            self.direction = Direction.DOWN
        else:
            self.direction = Direction.UP

        if self.direction == Direction.UP:
            if self.up_stops[0] == self.current_floor:
                heapq.heappop(self.up_stops)
                print(f"Elevator {self.id} stopped at {self.current_floor}")
            else:
                self.current_floor += 1
        else:
            if -self.down_stops[0] == self.current_floor:
                heapq.heappop(self.down_stops)
                print(f"Elevator {self.id} stopped at {self.current_floor}")
            else:
                self.current_floor -= 1

# ----- strategy: which lift serves a hall call? -----
class ElevatorSelectionStrategy(ABC):
    @abstractmethod
    def select(self, elevators: list[Elevator], req: Request) -> Elevator: ...

class NearestElevatorStrategy(ElevatorSelectionStrategy):
    def select(self, elevators, req):
        # "find the minimum in an array" - smallest distance wins
        return min(elevators, key=lambda e: abs(e.current_floor - req.floor_id))

class ElevatorSystem:
    def __init__(self, count: int, strategy: ElevatorSelectionStrategy):
        self.strategy = strategy
        self.elevators = [Elevator(i) for i in range(count)]
        self.threads = [threading.Thread(target=e.run, daemon=True) for e in self.elevators]
        for t in self.threads:             # keep the threads so we can stop/join them later
            t.start()

    def request_elevator(self, req: Request) -> Elevator:
        """Hall button: system picks the best lift, then hands it the request."""
        lift = self.strategy.select(self.elevators, req)
        print(f"Assigning elevator {lift.id} for floor {req.floor_id}")
        lift.add_request(req)
        return lift

    def shutdown(self) -> None:
        for e in self.elevators:
            e.stop()
        for t in self.threads:
            t.join()

# usage
system = ElevatorSystem(2, NearestElevatorStrategy())
lift = system.request_elevator(Request(5))   # someone presses the hall button on floor 5
time.sleep(0.4)
lift.add_request(Request(2))                 # inside the cabin they press 2 (no selection needed)
time.sleep(0.4)
system.shutdown()
# Assigning elevator 0 for floor 5
# Elevator 0 stopped at 5
# Elevator 0 stopped at 2