Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from datetime import datetime, timedelta
3from typing import Optional
4import uuid
5import time
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11usage_records = {}
12next_user_id = 1
13next_usage_id = 1
14 
15def get_current_user(authorization: Optional[str] = Header(None)):
16    if not authorization:
17        raise HTTPException(401, "No auth token")
18    token = authorization.replace("Bearer ", "")
19    if token not in tokens:
20        raise HTTPException(401, "Invalid token")
21    return tokens[token]
22 
23@app.post("/signup")
24def signup(username: str, password: str):
25    global next_user_id
26    if username in users:
27        raise HTTPException(400, "User exists")
28    user_id = next_user_id
29    next_user_id += 1
30    users[username] = {"id": user_id, "password": password}
31    return {"id": user_id, "username": username}
32 
33@app.post("/login")
34def login(username: str, password: str):
35    if username not in users or users[username]["password"] != password:
36        raise HTTPException(401, "Bad credentials")
37    token = str(uuid.uuid4())
38    tokens[token] = users[username]["id"]
39    return {"token": token}
40 
41@app.get("/usage/{usage_id}")
42def get_usage(usage_id: int, authorization: Optional[str] = Header(None)):
43    user_id = get_current_user(authorization)
44    if usage_id not in usage_records:
45        raise HTTPException(404, "Not found")
46    return usage_records[usage_id]
47 
48@app.post("/usage")
49def create_usage(gallons: float, authorization: Optional[str] = Header(None)):
50    global next_usage_id
51    user_id = get_current_user(authorization)
52    usage_id = next_usage_id
53    next_usage_id += 1
54    usage_records[usage_id] = {
55        "id": usage_id,
56        "user_id": user_id,
57        "gallons": gallons,
58        "timestamp": datetime.now().isoformat()
59    }
60    return usage_records[usage_id]
61 
62@app.get("/usage")
63def get_daily_usage(authorization: Optional[str] = Header(None)):
64    user_id = get_current_user(authorization)
65    today = datetime.now().date()
66    today_records = [r for r in usage_records.values() if r["user_id"] == user_id and datetime.fromisoformat(r["timestamp"]).date() == today]
67    total_daily = sum(r["gallons"] for r in today_records)
68 
69    # weekly average
70    week_ago = datetime.now() - timedelta(days=7)
71    week_records = [r for r in usage_records.values() if r["user_id"] == user_id and datetime.fromisoformat(r["timestamp"]) >= week_ago]
72    weekly_avg = sum(r["gallons"] for r in week_records) / 7 if week_records else 0
73 
74    # leak detection: check if any flow > 3 hours straight
75    leak_detected = False
76    user_records_sorted = sorted([r for r in usage_records.values() if r["user_id"] == user_id], key=lambda x: x["timestamp"])
77    if len(user_records_sorted) >= 2:
78        for i in range(len(user_records_sorted) - 1):
79            t1 = datetime.fromisoformat(user_records_sorted[i]["timestamp"])
80            t2 = datetime.fromisoformat(user_records_sorted[i+1]["timestamp"])
81            if (t2 - t1).total_seconds() > 3 * 3600 and user_records_sorted[i]["gallons"] > 0:
82                leak_detected = True
83                break
84 
85    return {
86        "daily_gallons": total_daily,
87        "weekly_average_gallons": round(weekly_avg, 2),
88        "leak_detected": leak_detected
89    }

1fastapi
2uvicorn