Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3from typing import Optional, List
4import math
5import random
6import string
7import datetime
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13orders = {}
14drivers = {}
15driver_schedules = {}
16next_user_id = 1
17next_order_id = 1
18next_driver_id = 1
19next_token_id = 1
20 
21def generate_token():
22    return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
23 
24def get_current_user(authorization: Optional[str] = Header(None)):
25    if not authorization:
26        raise HTTPException(status_code=401, detail="Missing auth token")
27    token = authorization.replace("Bearer ", "")
28    user_id = tokens.get(token)
29    if not user_id:
30        raise HTTPException(status_code=401, detail="Invalid auth token")
31    return user_id
32 
33class SignupRequest(BaseModel):
34    username: str
35    password: str
36 
37class LoginRequest(BaseModel):
38    username: str
39    password: str
40 
41class Address(BaseModel):
42    street: str
43    city: str
44    lat: float
45    lng: float
46 
47class BatchRouteRequest(BaseModel):
48    addresses: List[Address]
49 
50class OrderCreate(BaseModel):
51    description: str
52    address: Address
53 
54@app.post("/signup")
55def signup(req: SignupRequest):
56    global next_user_id
57    user_id = next_user_id
58    next_user_id += 1
59    users[user_id] = {"username": req.username, "password": req.password}
60    token = generate_token()
61    tokens[token] = user_id
62    return {"user_id": user_id, "token": token}
63 
64@app.post("/login")
65def login(req: LoginRequest):
66    for uid, u in users.items():
67        if u["username"] == req.username and u["password"] == req.password:
68            token = generate_token()
69            tokens[token] = uid
70            return {"user_id": uid, "token": token}
71    raise HTTPException(status_code=401, detail="Invalid credentials")
72 
73@app.get("/orders/{order_id}")
74def get_order(order_id: int, authorization: Optional[str] = Header(None)):
75    user_id = get_current_user(authorization)
76    order = orders.get(order_id)
77    if not order:
78        raise HTTPException(status_code=404, detail="Order not found")
79    return order
80 
81@app.post("/orders")
82def create_order(req: OrderCreate, authorization: Optional[str] = Header(None)):
83    user_id = get_current_user(authorization)
84    global next_order_id
85    order_id = next_order_id
86    next_order_id += 1
87    orders[order_id] = {
88        "id": order_id,
89        "user_id": user_id,
90        "description": req.description,
91        "address": req.address.dict(),
92        "status": "pending"
93    }
94    return orders[order_id]
95 
96@app.post("/routes/batch")
97def optimize_route(req: BatchRouteRequest, authorization: Optional[str] = Header(None)):
98    user_id = get_current_user(authorization)
99    if not req.addresses:
100        raise HTTPException(status_code=400, detail="No addresses provided")
101 
102    # Simple nearest-neighbor TSP heuristic
103    addresses = req.addresses.copy()
104    sequence = []
105    total_distance = 0.0
106 
107    # Start from first address
108    current = addresses.pop(0)
109    sequence.append(current)
110 
111    while addresses:
112        nearest = None
113        nearest_dist = float('inf')
114        for addr in addresses:
115            dist = math.sqrt((addr.lat - current.lat)**2 + (addr.lng - current.lng)**2)
116            if dist < nearest_dist:
117                nearest_dist = dist
118                nearest = addr
119        total_distance += nearest_dist
120        current = nearest
121        sequence.append(current)
122        addresses.remove(nearest)
123 
124    # Average speed 30 km/h, convert distance (in lat/lng units ~111km per degree)
125    estimated_time_hours = (total_distance * 111) / 30
126    estimated_time_minutes = estimated_time_hours * 60
127 
128    return {
129        "optimized_sequence": [s.dict() for s in sequence],
130        "total_distance_km": round(total_distance * 111, 2),
131        "estimated_time_minutes": round(estimated_time_minutes, 2)
132    }
133 
134@app.post("/drivers")
135def create_driver(name: str, authorization: Optional[str] = Header(None)):
136    user_id = get_current_user(authorization)
137    global next_driver_id
138    driver_id = next_driver_id
139    next_driver_id += 1
140    drivers[driver_id] = {
141        "id": driver_id,
142        "name": name,
143        "user_id": user_id
144    }
145    # Initialize empty schedule for today
146    today = datetime.date.today().isoformat()
147    driver_schedules[driver_id] = {today: []}
148    return drivers[driver_id]
149 
150@app.get("/drivers/{driver_id}/schedule")
151def get_driver_schedule(driver_id: int, authorization: Optional[str] = Header(None)):
152    user_id = get_current_user(authorization)
153    driver = drivers.get(driver_id)
154    if not driver:
155        raise HTTPException(status_code=404, detail="Driver not found")
156    today = datetime.date.today().isoformat()
157    schedule = driver_schedules.get(driver_id, {}).get(today, [])
158    return {"driver_id": driver_id, "date": today, "stops": schedule}

1fastapi
2uvicorn