server.py

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from contextlib import asynccontextmanager
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from collections import deque
from datetime import datetime
from io import BytesIO
from dotenv import load_dotenv
from google.cloud import storage
import asyncio, base64, calendar, json, os, threading, time
import cv2, numpy as np, mediapipe as mp, requests

load_dotenv()

# ── Telegram ──────────────────────────────────────────────────────────────────
TELEGRAM_TOKEN   = os.getenv("TELEGRAM_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")

def send_telegram_alert(message: str):
    if not TELEGRAM_TOKEN or not TELEGRAM_CHAT_ID:
        print("[TELEGRAM] Skipped — TOKEN or CHAT_ID not set")
        return
    try:
        resp = requests.post(
            f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage",
            data={"chat_id": TELEGRAM_CHAT_ID, "text": message},
            timeout=10,
        )
        print("[TELEGRAM] Sent" if resp.ok else f"[TELEGRAM] Failed: {resp.status_code}")
    except Exception as e:
        print(f"[TELEGRAM] Error: {e}")

# ── App & MediaPipe ───────────────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
    asyncio.create_task(ai_inference_worker())
    yield

app = FastAPI(lifespan=lifespan)
app.mount("/custom", StaticFiles(directory="custom"), name="custom")

mp_pose    = mp.solutions.pose
mp_drawing = mp.solutions.drawing_utils
pose       = mp_pose.Pose(min_detection_confidence=0.6, min_tracking_confidence=0.6)

# ── Fall log persistence ──────────────────────────────────────────────────────
FALL_LOG_FILE = "fall_log.json"

def _load_fall_log() -> list[datetime]:
    try:
        with open(FALL_LOG_FILE, encoding="utf-8") as f:
            return [datetime.fromisoformat(ts) for ts in json.load(f)]
    except (FileNotFoundError, json.JSONDecodeError):
        return []

def _save_fall_log(log: list[datetime]):
    with open(FALL_LOG_FILE, "w", encoding="utf-8") as f:
        json.dump([ts.isoformat() for ts in log], f, indent=2)

# ── State ─────────────────────────────────────────────────────────────────────
latest_frame:   bytes | None    = None
fall_frame:     bytes | None    = None
last_fall_time: float           = 0.0
fall_counter:   int             = 0
body_angle:     str             = "front"
metrics_data:   dict            = {"cpu": 0, "memory": 0}
hip_history:    deque           = deque(maxlen=4)
fall_log:       list[datetime]  = _load_fall_log()
alarm_active:    bool             = False
pi_ws_ref:       WebSocket | None = None
last_fall_label: str              = ""

print(f"[FALL LOG] Loaded {len(fall_log)} past events")

frame_queue: asyncio.Queue = asyncio.Queue(maxsize=1)
web_clients: set           = set()

FALL_DETECTION_FRAMES = 5
FALL_COOLDOWN         = 5

# ── Storage ───────────────────────────────────────────────────────────────────
def log_fall_event(image_bytes: bytes, timestamp: float, label: str):
    try:
        bucket = storage.Client().bucket("fall-log-data")
        bucket.blob(f"fall_events/{label}.jpg").upload_from_string(image_bytes, content_type="image/jpeg")
        bucket.blob(f"fall_events/{label}.json").upload_from_string(
            json.dumps({"event": "fall_detected", "timestamp": label}, indent=2),
            content_type="application/json",
        )
        print(f"[GCS] Uploaded {label}")
    except Exception:
        os.makedirs("fall_events", exist_ok=True)
        with open(f"fall_events/{label}.jpg", "wb") as f:
            f.write(image_bytes)
        print(f"[LOCAL] Saved fall_events/{label}.jpg")

# ── Fall detection ────────────────────────────────────────────────────────────
def determine_body_orientation(lm) -> str:
    shoulder_wide = abs(lm[11][0] - lm[12][0])
    s_h_high = abs((lm[23][1] + lm[24][1] - lm[11][1] - lm[12][1]) / 2)
    rate = shoulder_wide / s_h_high if s_h_high > 0 else 0
    if rate < 0.2: return "sideway whole"
    if rate < 0.4: return "sideway slight"
    return "front"

def detect_fall(lm) -> tuple[bool, str]:
    sh_y  = (lm[11][1] + lm[12][1]) / 2
    sh_x  = (lm[11][0] + lm[12][0]) / 2
    hip_y = (lm[23][1] + lm[24][1]) / 2
    hip_x = (lm[23][0] + lm[24][0]) / 2
    spine_len = np.sqrt((sh_x - hip_x)**2 + (sh_y - hip_y)**2)

    angle = np.degrees(np.arccos(np.clip(abs(sh_y - hip_y) / spine_len, 0, 1))) if spine_len > 0.01 else 0.0
    nose_near = lm[0][1] > hip_y - 0.05

    hip_history.append(hip_y)
    velocity = (hip_history[-1] - hip_history[0]) if len(hip_history) >= 2 else 0.0

    # Bounding box aspect ratio — quan trọng với camera góc cao
    pts = [(lm[i][0], lm[i][1]) for i in range(33) if lm[i][3] > 0.3]
    bbox_fallen = False
    if len(pts) >= 6:
        xs, ys = zip(*pts)
        w = max(xs) - min(xs)
        h = max(ys) - min(ys)
        if w > 0.02:
            aspect = h / w  # < 1.0 = nằm ngang = ngã
            bbox_fallen = aspect < 0.8 and velocity > 0.010

    is_fall = (
        angle > 65
        or (angle > 50 and nose_near and velocity > 0.015)
        or (velocity > 0.030 and angle > 35)
        or bbox_fallen
    )
    status = f"Fall! {angle:.0f}deg v={velocity:.3f}" if is_fall else f"Normal {angle:.0f}deg"
    return is_fall, status

# ── Frame processing ──────────────────────────────────────────────────────────
def process_frame(img_bytes: bytes) -> tuple[bytes | None, bool]:
    global fall_counter, last_fall_time, fall_frame, body_angle, alarm_active, last_fall_label

    np_img = np.frombuffer(img_bytes, np.uint8)
    img    = cv2.imdecode(np_img, cv2.IMREAD_COLOR)
    if img is None:
        return None, False

    results      = pose.process(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
    is_falling   = False
    current_time = time.time()
    status       = "No pose"

    if results.pose_landmarks:
        mp_drawing.draw_landmarks(img, results.pose_landmarks, mp_pose.POSE_CONNECTIONS)
        lm = [(p.x, p.y, p.z, p.visibility) for p in results.pose_landmarks.landmark]
        body_angle   = determine_body_orientation(lm)
        is_fall, status = detect_fall(lm)

        if is_fall:
            fall_counter += 1
            if fall_counter >= FALL_DETECTION_FRAMES and current_time - last_fall_time > FALL_COOLDOWN:
                is_falling      = True
                alarm_active    = True
                last_fall_time  = current_time
                fall_counter    = 0
                last_fall_label = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime(current_time))
                fall_log.append(datetime.now())
                _save_fall_log(fall_log)
                print(f"[FALL] {status}")
                _, buf = cv2.imencode(".jpg", img)
                fall_frame = buf.tobytes()
                threading.Thread(target=log_fall_event,      args=(fall_frame, current_time, last_fall_label), daemon=True).start()
                threading.Thread(target=send_telegram_alert, args=("⚠️ CẢNH BÁO: Phát hiện ngã! Vui lòng kiểm tra ngay.",), daemon=True).start()
        else:
            fall_counter = max(0, fall_counter - 1)

    cv2.rectangle(img, (0, 0), (225, 130), (245, 117, 16), -1)
    cv2.putText(img, "Fall Counter",     (15,  12), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0),       1, cv2.LINE_AA)
    cv2.putText(img, str(fall_counter),  (10,  65), cv2.FONT_HERSHEY_SIMPLEX, 2.0, (255, 255, 255), 2, cv2.LINE_AA)
    cv2.putText(img, body_angle,         (10, 110), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2, cv2.LINE_AA)
    cv2.putText(img, status,             (10, 150), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0),     2, cv2.LINE_AA)

    _, jpeg = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 65])
    return jpeg.tobytes(), is_falling

# ── Background AI worker ──────────────────────────────────────────────────────
async def ai_inference_worker():
    global latest_frame
    while True:
        try:
            img_bytes, pi_ws      = await frame_queue.get()
            processed, is_falling = await asyncio.to_thread(process_frame, img_bytes)
            if not processed:
                continue

            latest_frame = processed

            if is_falling and pi_ws:
                try:
                    await pi_ws.send_json({"event": "FALL"})
                except Exception as e:
                    print(f"[WS] Send to Pi failed: {e}")

            if web_clients:
                msg  = json.dumps({"image": base64.b64encode(processed).decode(), "alarm": alarm_active})
                dead = set()
                for ws in web_clients:
                    try:    await ws.send_text(msg)
                    except: dead.add(ws)
                web_clients.difference_update(dead)
        except Exception as e:
            print(f"[WORKER] {e}")

# ── WebSocket endpoints ───────────────────────────────────────────────────────
@app.websocket("/ws/pi")
async def ws_pi(websocket: WebSocket):
    global pi_ws_ref
    await websocket.accept()
    pi_ws_ref = websocket
    print("[WS] Pi connected")
    try:
        while True:
            data = await websocket.receive_bytes()
            if frame_queue.full():
                try: frame_queue.get_nowait()
                except asyncio.QueueEmpty: pass
            await frame_queue.put((data, websocket))
    except WebSocketDisconnect:
        pi_ws_ref = None
        print("[WS] Pi disconnected")

@app.websocket("/ws/web")
async def ws_web(websocket: WebSocket):
    await websocket.accept()
    web_clients.add(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        web_clients.discard(websocket)

# ── HTTP endpoints ────────────────────────────────────────────────────────────
def _html(path: str) -> str:
    return open(path, encoding="utf-8").read()

@app.get("/",          response_class=HTMLResponse)
async def index():     return _html("templates/index.html")

@app.get("/login",     response_class=HTMLResponse)
async def login():     return _html("templates/login.html")

@app.get("/camera",    response_class=HTMLResponse)
async def camera():    return _html("templates/camera.html")

@app.get("/chart",     response_class=HTMLResponse)
async def chart():     return _html("templates/chart.html")

@app.get("/fallchart", response_class=HTMLResponse)
async def fallchart(): return _html("templates/fallchart.html")

@app.get("/trigger_feed")
async def trigger_feed():
    if fall_frame:
        return StreamingResponse(BytesIO(fall_frame), media_type="image/jpeg")
    blank = np.zeros((200, 300, 3), dtype=np.uint8)
    _, jpeg = cv2.imencode(".jpg", blank)
    return StreamingResponse(BytesIO(jpeg.tobytes()), media_type="image/jpeg")

@app.post("/metrics")
async def receive_metrics(data: dict):
    metrics_data.update(data)
    return {"status": "received"}

@app.get("/get_metrics")
async def get_metrics():
    return metrics_data

@app.post("/reset_alarm")
async def reset_alarm():
    global alarm_active, fall_frame, fall_log, last_fall_label
    alarm_active = False
    fall_frame   = None

    if fall_log:
        fall_log.pop()
        _save_fall_log(fall_log)

    if last_fall_label:
        for ext in (".jpg", ".json"):
            path = f"fall_events/{last_fall_label}{ext}"
            try:
                os.remove(path)
            except FileNotFoundError:
                pass
        print(f"[ALARM] False alarm cleared — {last_fall_label}")
        last_fall_label = ""

    if pi_ws_ref:
        try:
            await pi_ws_ref.send_json({"event": "RESET"})
        except Exception as e:
            print(f"[ALARM] Reset send to Pi failed: {e}")

    return {"status": "ok"}

@app.get("/fall_stats")
async def fall_stats(month: str | None = None):
    now = datetime.now()

    # Parse requested month, default to current
    try:
        target = datetime.strptime(month, "%Y-%m") if month else now
    except ValueError:
        target = now
    target_month = (target.year, target.month)

    hourly        = {f"{h:02d}:00": 0 for h in range(24)}
    days_in_month = calendar.monthrange(*target_month)[1]
    daily         = {str(d): 0 for d in range(1, days_in_month + 1)}

    month_total = 0
    for ts in fall_log:
        if (ts.year, ts.month) == target_month:
            daily[str(ts.day)] += 1
            month_total += 1
            # Hourly chỉ show khi xem tháng hiện tại (ngày hôm nay)
            if ts.date() == now.date():
                hourly[f"{ts.hour:02d}:00"] += 1

    # Danh sách tháng có dữ liệu
    available = sorted({f"{ts.year:04d}-{ts.month:02d}" for ts in fall_log}, reverse=True)
    if not available:
        available = [now.strftime("%Y-%m")]

    return {
        "hourly":    hourly,
        "daily":     daily,
        "total":     len(fall_log),
        "month_total": month_total,
        "months":    available,
        "current_month": f"{target_month[0]:04d}-{target_month[1]:02d}",
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=False)