Food Delivery (Swiggy)

  1  from __future__ import annotations
  2  from abc import ABC, abstractmethod
  3  from enum import Enum
  4  from dataclasses import dataclass, field
  5  from typing import Protocol
  6  from threading import Lock
  7  import uuid
  8  import math
  9  
 10  
 11  class OrderStatus(Enum):
 12      PLACED = "placed"
 13      CONFIRMED = "confirmed"
 14      PREPARING = "preparing"
 15      OUT_FOR_DELIVERY = "out_for_delivery"
 16      DELIVERED = "delivered"
 17      CANCELLED = "cancelled"
 18  
 19  
 20  VALID_TRANSITIONS: dict[OrderStatus, set[OrderStatus]] = {
 21      OrderStatus.PLACED: {OrderStatus.CONFIRMED, OrderStatus.CANCELLED},
 22      OrderStatus.CONFIRMED: {OrderStatus.PREPARING, OrderStatus.CANCELLED},
 23      OrderStatus.PREPARING: {OrderStatus.OUT_FOR_DELIVERY, OrderStatus.CANCELLED},
 24      OrderStatus.OUT_FOR_DELIVERY: {OrderStatus.DELIVERED},
 25      OrderStatus.DELIVERED: set(),
 26      OrderStatus.CANCELLED: set(),
 27  }
 28  
 29  
 30  @dataclass
 31  class Location:
 32      lat: float
 33      lng: float
 34  
 35      def distance_to(self, other: "Location") -> float:
 36          """Haversine distance in km."""
 37          R = 6371
 38          lat1, lat2 = math.radians(self.lat), math.radians(other.lat)
 39          dlat = math.radians(other.lat - self.lat)
 40          dlng = math.radians(other.lng - self.lng)
 41          a = (math.sin(dlat / 2) ** 2
 42               + math.cos(lat1) * math.cos(lat2) * math.sin(dlng / 2) ** 2)
 43          return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
 44  
 45  
 46  @dataclass
 47  class MenuItem:
 48      name: str
 49      price: float
 50      available: bool = True
 51  
 52  
 53  @dataclass
 54  class OrderItem:
 55      menu_item: MenuItem
 56      quantity: int
 57  
 58      @property
 59      def subtotal(self) -> float:
 60          return self.menu_item.price * self.quantity
 61  
 62  
 63  @dataclass
 64  class Customer:
 65      id: str
 66      name: str
 67      address: Location
 68  
 69  
 70  class Restaurant:
 71      def __init__(self, restaurant_id: str, name: str, location: Location,
 72                   menu: list[MenuItem]):
 73          self.id = restaurant_id
 74          self.name = name
 75          self.location = location
 76          self._menu = {item.name: item for item in menu}
 77  
 78      def get_item(self, name: str) -> MenuItem | None:
 79          return self._menu.get(name)
 80  
 81      def get_available_items(self) -> list[MenuItem]:
 82          return [item for item in self._menu.values() if item.available]
 83  
 84  
 85  class DeliveryAgent:
 86      def __init__(self, agent_id: str, name: str, location: Location,
 87                   max_load: int = 3):
 88          self.id = agent_id
 89          self.name = name
 90          self.location = location
 91          self.current_load = 0
 92          self.max_load = max_load
 93          self.rating = 4.5
 94  
 95      def is_available(self) -> bool:
 96          return self.current_load < self.max_load
 97  
 98      def assign_order(self) -> None:
 99          if not self.is_available():
100              raise RuntimeError(f"Agent {self.name} is at max capacity")
101          self.current_load += 1
102  
103      def complete_order(self) -> None:
104          self.current_load = max(0, self.current_load - 1)
105  
106  
107  class OrderObserver(ABC):
108      @abstractmethod
109      def on_status_change(self, order_id: str, customer_name: str,
110                           restaurant_name: str, agent_name: str | None,
111                           old_status: OrderStatus,
112                           new_status: OrderStatus) -> None: ...
113  
114  
115  class ConsoleNotifier(OrderObserver):
116      def on_status_change(self, order_id: str, customer_name: str,
117                           restaurant_name: str, agent_name: str | None,
118                           old_status: OrderStatus,
119                           new_status: OrderStatus) -> None:
120          agent_info = f" | agent: {agent_name}" if agent_name else ""
121          print(f"  [ORDER {order_id}] {old_status.value} -> {new_status.value} | "
122                f"{customer_name} from {restaurant_name}{agent_info}")
123  
124  
125  class Order:
126      def __init__(self, order_id: str, customer: Customer,
127                   restaurant: Restaurant, items: list[OrderItem]):
128          self.id = order_id
129          self.customer = customer
130          self.restaurant = restaurant
131          self.items = items
132          self.agent: DeliveryAgent | None = None
133          self._status = OrderStatus.PLACED
134          self.total = sum(item.subtotal for item in items)
135          self._observers: list[OrderObserver] = []
136  
137      @property
138      def status(self) -> OrderStatus:
139          return self._status
140  
141      def add_observer(self, observer: OrderObserver) -> None:
142          self._observers.append(observer)
143  
144      def _transition(self, new_status: OrderStatus) -> None:
145          if new_status not in VALID_TRANSITIONS[self._status]:
146              raise ValueError(
147                  f"Cannot transition from {self._status.value} to "
148                  f"{new_status.value}")
149          old = self._status
150          self._status = new_status
151          agent_name = self.agent.name if self.agent else None
152          for obs in self._observers:
153              obs.on_status_change(
154                  self.id, self.customer.name, self.restaurant.name,
155                  agent_name, old, new_status)
156  
157      def confirm(self) -> None:
158          self._transition(OrderStatus.CONFIRMED)
159  
160      def start_preparing(self) -> None:
161          self._transition(OrderStatus.PREPARING)
162  
163      def dispatch(self, agent: DeliveryAgent) -> None:
164          agent.assign_order()
165          self.agent = agent
166          self._transition(OrderStatus.OUT_FOR_DELIVERY)
167  
168      def deliver(self) -> None:
169          self._transition(OrderStatus.DELIVERED)
170          if self.agent:
171              self.agent.complete_order()
172  
173      def cancel(self) -> None:
174          self._transition(OrderStatus.CANCELLED)
175          if self.agent:
176              self.agent.complete_order()
177  
178  
179  class AssignmentStrategy(Protocol):
180      def assign_agent(self, agents: list[DeliveryAgent],
181                       restaurant: Restaurant) -> DeliveryAgent | None: ...
182  
183  
184  class NearestAgentStrategy:
185      """Pick the closest available agent to the restaurant."""
186      def assign_agent(self, agents: list[DeliveryAgent],
187                       restaurant: Restaurant) -> DeliveryAgent | None:
188          available = [a for a in agents if a.is_available()]
189          if not available:
190              return None
191          return min(available,
192                     key=lambda a: a.location.distance_to(restaurant.location))
193  
194  
195  class LoadBalancedStrategy:
196      """Pick the available agent with the fewest current orders."""
197      def assign_agent(self, agents: list[DeliveryAgent],
198                       restaurant: Restaurant) -> DeliveryAgent | None:
199          available = [a for a in agents if a.is_available()]
200          if not available:
201              return None
202          return min(available, key=lambda a: a.current_load)
203  
204  
205  class DeliveryPlatform:
206      def __init__(self, assignment: AssignmentStrategy | None = None):
207          self._restaurants: dict[str, Restaurant] = {}
208          self._agents: list[DeliveryAgent] = []
209          self._orders: dict[str, Order] = {}
210          self._assignment = assignment or NearestAgentStrategy()
211          self._notifier = ConsoleNotifier()
212          self._lock = Lock()
213  
214      def register_restaurant(self, restaurant: Restaurant) -> None:
215          self._restaurants[restaurant.id] = restaurant
216  
217      def register_agent(self, agent: DeliveryAgent) -> None:
218          self._agents.append(agent)
219  
220      def set_assignment_strategy(self, strategy: AssignmentStrategy) -> None:
221          self._assignment = strategy
222  
223      def place_order(self, customer: Customer, restaurant_id: str,
224                      item_requests: list[tuple[str, int]]) -> Order:
225          """Place an order. item_requests is a list of (item_name, quantity)."""
226          with self._lock:
227              restaurant = self._restaurants.get(restaurant_id)
228              if not restaurant:
229                  raise ValueError(f"Restaurant {restaurant_id} not found")
230  
231              order_items = []
232              for item_name, qty in item_requests:
233                  menu_item = restaurant.get_item(item_name)
234                  if menu_item is None:
235                      raise ValueError(f"Item '{item_name}' not on menu")
236                  if not menu_item.available:
237                      raise ValueError(f"Item '{item_name}' is unavailable")
238                  order_items.append(OrderItem(menu_item, qty))
239  
240              order_id = str(uuid.uuid4())[:8]
241              order = Order(order_id, customer, restaurant, order_items)
242              order.add_observer(self._notifier)
243              self._orders[order_id] = order
244              return order
245  
246      def confirm_order(self, order_id: str) -> None:
247          """Restaurant confirms it can fulfill the order."""
248          self._orders[order_id].confirm()
249  
250      def mark_preparing(self, order_id: str) -> None:
251          self._orders[order_id].start_preparing()
252  
253      def dispatch_order(self, order_id: str) -> None:
254          """Assign an agent and send the order out for delivery."""
255          with self._lock:
256              order = self._orders[order_id]
257              agent = self._assignment.assign_agent(
258                  self._agents, order.restaurant)
259              if agent is None:
260                  raise RuntimeError("No delivery agents available")
261              order.dispatch(agent)
262  
263      def complete_delivery(self, order_id: str) -> None:
264          with self._lock:
265              self._orders[order_id].deliver()
266  
267      def cancel_order(self, order_id: str) -> None:
268          with self._lock:
269              self._orders[order_id].cancel()
270  
271  
272  if __name__ == "__main__":
273      platform = DeliveryPlatform(assignment=NearestAgentStrategy())
274  
275      # Set up a restaurant
276      biryani_house = Restaurant("r1", "Biryani House", Location(12.9716, 77.5946), [
277          MenuItem("Chicken Biryani", 250.0),
278          MenuItem("Paneer Biryani", 220.0),
279          MenuItem("Raita", 40.0),
280          MenuItem("Gulab Jamun", 60.0),
281      ])
282      platform.register_restaurant(biryani_house)
283  
284      # Set up delivery agents
285      platform.register_agent(DeliveryAgent("a1", "Rahul", Location(12.9720, 77.5950)))
286      platform.register_agent(DeliveryAgent("a2", "Meena", Location(12.9600, 77.6100)))
287      platform.register_agent(DeliveryAgent("a3", "Vijay", Location(12.9800, 77.5800)))
288  
289      # Customer places an order
290      customer = Customer("c1", "Priya", Location(12.9352, 77.6245))
291  
292      print("=== Placing order ===")
293      order = platform.place_order(
294          customer, "r1",
295          [("Chicken Biryani", 2), ("Raita", 1)])
296      print(f"  Order {order.id}: Rs {order.total}")
297  
298      print("\n=== Restaurant confirms ===")
299      platform.confirm_order(order.id)
300  
301      print("\n=== Kitchen starts preparing ===")
302      platform.mark_preparing(order.id)
303  
304      print("\n=== Dispatching delivery agent ===")
305      platform.dispatch_order(order.id)
306      print(f"  Assigned to: {order.agent.name}")
307  
308      print("\n=== Delivered ===")
309      platform.complete_delivery(order.id)
310  
311      # Second order with load balancing
312      print("\n=== Switching to load-balanced assignment ===")
313      platform.set_assignment_strategy(LoadBalancedStrategy())
314      customer2 = Customer("c2", "Amit", Location(12.9400, 77.6000))
315      order2 = platform.place_order(
316          customer2, "r1",
317          [("Paneer Biryani", 1), ("Gulab Jamun", 2)])
318      platform.confirm_order(order2.id)
319      platform.mark_preparing(order2.id)
320      platform.dispatch_order(order2.id)
321      print(f"  Order {order2.id} assigned to: {order2.agent.name} "
322            f"(load: {order2.agent.current_load}/{order2.agent.max_load})")
323      platform.complete_delivery(order2.id)
324  
325      print("\n=== Invalid transition test ===")
326      try:
327          platform.dispatch_order(order.id)  # already delivered
328      except ValueError as e:
329          print(f"  Correctly rejected: {e}")
330  
331      print("\n=== Cancellation test ===")
332      order3 = platform.place_order(
333          customer, "r1", [("Gulab Jamun", 3)])
334      print(f"  Placed order {order3.id}")
335      platform.cancel_order(order3.id)
336      print(f"  Status after cancel: {order3.status.value}")