From b46f5d0f8c125e941633960fbd6eea0ac7a08486 Mon Sep 17 00:00:00 2001 From: AI-WonYJ Date: Sun, 12 Jul 2026 17:51:29 +0900 Subject: [PATCH] =?UTF-8?q?refactor(#14):=20FastAPI=20backend=20=EB=9D=BC?= =?UTF-8?q?=EC=9A=B0=ED=84=B0,=20=EC=84=9C=EB=B9=84=EC=8A=A4,=20DB=20?= =?UTF-8?q?=EB=AA=A8=EB=93=88=20=EB=B6=84=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - DB connection pool 및 공통 connection helper를 database.py로 분리 - YOLO 이미지 디코딩과 추론 로직을 inference_service.py로 분리 - 통계 조회 및 추론 결과 저장 로직을 stats_service.py로 분리 - public API와 device API를 각각 전용 router 모듈로 분리 - main.py를 FastAPI 앱 초기화와 라우터 등록 중심으로 정리 - 기존 API 경로와 응답 구조 유지 --- backend/database.py | 47 +++ backend/main.py | 530 +------------------------- backend/routers/__init__.py | 0 backend/routers/device.py | 105 +++++ backend/routers/public.py | 33 ++ backend/services/__init__.py | 0 backend/services/inference_service.py | 37 ++ backend/services/stats_service.py | 362 ++++++++++++++++++ 8 files changed, 593 insertions(+), 521 deletions(-) create mode 100644 backend/database.py create mode 100644 backend/routers/__init__.py create mode 100644 backend/routers/device.py create mode 100644 backend/routers/public.py create mode 100644 backend/services/__init__.py create mode 100644 backend/services/inference_service.py create mode 100644 backend/services/stats_service.py diff --git a/backend/database.py b/backend/database.py new file mode 100644 index 0000000..8c5d805 --- /dev/null +++ b/backend/database.py @@ -0,0 +1,47 @@ +import os + +from dotenv import load_dotenv +from psycopg_pool import ConnectionPool + +load_dotenv() + +_db_pool: ConnectionPool | None = None + + +def open_database_pool() -> None: + """애플리케이션 시작 시 PostgreSQL connection pool을 생성합니다.""" + global _db_pool + + if _db_pool is not None: + return + + database_url = os.getenv("DATABASE_URL") + + if not database_url: + raise RuntimeError("DATABASE_URL is not set") + + _db_pool = ConnectionPool( + conninfo=database_url, + min_size=1, + max_size=10, + open=True, + ) + + +def close_database_pool() -> None: + """애플리케이션 종료 시 PostgreSQL connection pool을 닫습니다.""" + global _db_pool + + if _db_pool is None: + return + + _db_pool.close() + _db_pool = None + + +def get_db_connection(): + """생성된 connection pool에서 DB connection을 반환합니다.""" + if _db_pool is None: + raise RuntimeError("Database connection pool is not initialized") + + return _db_pool.connection() diff --git a/backend/main.py b/backend/main.py index 35b4b3c..fc49b01 100644 --- a/backend/main.py +++ b/backend/main.py @@ -1,45 +1,23 @@ -import os -import time from contextlib import asynccontextmanager -from datetime import datetime -from decimal import Decimal -from typing import Annotated -import cv2 -import numpy as np -from dotenv import load_dotenv -from fastapi import FastAPI, File, UploadFile +from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles -from psycopg_pool import ConnectionPool -from ultralytics import YOLO -load_dotenv() - -DATABASE_URL = os.getenv("DATABASE_URL") - -db_pool = None +from database import close_database_pool, open_database_pool +from routers.device import router as device_router +from routers.public import router as public_router @asynccontextmanager -async def lifespan(app: FastAPI): - global db_pool - - if not DATABASE_URL: - raise RuntimeError("DATABASE_URL is not set") - - db_pool = ConnectionPool( - conninfo=DATABASE_URL, - min_size=1, - max_size=10, - open=True, - ) +async def lifespan(_: FastAPI): + open_database_pool() try: yield finally: - db_pool.close() + close_database_pool() app = FastAPI(title="SSCCounter API", version="2.1", lifespan=lifespan) @@ -58,412 +36,12 @@ async def lifespan(app: FastAPI): allow_headers=["*"], ) -# YOLO 모델 로드 (yolo_inference.py에서 가져옴) -print("Loading YOLO model...") -model = YOLO("yolov8n.pt") -print("YOLO model loaded successfully.") - - -def infer_people_count(frame): - results = model.predict( - source=frame, conf=0.3, classes=[0], imgsz=416, verbose=False - ) - return (results[0].boxes.cls == 0).sum().item() - - -# --------------------------------------------- - - -def get_db_connection(): - if db_pool is None: - raise RuntimeError("Database connection pool is not initialized") - return db_pool.connection() - - -def to_float(value): - if value is None: - return 0.0 - if isinstance(value, Decimal): - return float(value) - return value - - -def get_current_status_from_db(): - with get_db_connection() as conn: - with conn.cursor() as cur: - cur.execute( - """ - SELECT - current_people_count, - updated_at, - today_max_people_count, - today_avg_people_count - FROM dashboard.current_status - WHERE id = 1 - AND updated_at >= date_trunc('day', now()) - AND updated_at < date_trunc('day', now()) + interval '1 day' - """ - ) - row = cur.fetchone() - - if row is None: - return { - "count": 0, - "updated_at": None, - "today_max_count": 0, - "today_avg_count": 0.0, - } - - return { - "count": row[0], - "updated_at": row[1].isoformat() if row[1] else None, - "today_max_count": row[2], - "today_avg_count": to_float(row[3]), - } - - -def save_failed_inference( - error_message: str, - trigger_type: str = "scheduled", - inference_time_ms: int | None = None, -): - with get_db_connection() as conn: - with conn.cursor() as cur: - cur.execute( - """ - INSERT INTO admin.capture_logs ( - people_count, - status, - trigger_type, - inference_time_ms, - error_message - ) - VALUES (%s, %s, %s, %s, %s) - """, - (None, "failed", trigger_type, inference_time_ms, error_message), - ) - - -def save_successful_inference( - people_count: int, - trigger_type: str = "scheduled", - inference_time_ms: int | None = None, -): - """ - YOLO 추론 성공 결과를 admin raw log와 dashboard summary tables에 반영합니다. - - 처리 순서: - 1. admin.capture_logs insert - 2. dashboard.current_status update - 3. dashboard.today_hourly_stats update - 4. dashboard.daily_stats update - 5. dashboard.weekday_hourly_stats update - """ - with get_db_connection() as conn: - with conn.cursor() as cur: - # 1. 관리자용 raw log 저장 - cur.execute( - """ - INSERT INTO admin.capture_logs ( - people_count, - status, - trigger_type, - inference_time_ms - ) - VALUES (%s, %s, %s, %s) - RETURNING measured_at - """, - (people_count, "success", trigger_type, inference_time_ms), - ) - measured_at = cur.fetchone()[0] - - # 2. dashboard.current_status 갱신 - cur.execute( - """ - WITH today_summary AS ( - SELECT - COALESCE(MAX(people_count), 0) AS today_max_people_count, - COALESCE( - ROUND(AVG(people_count)::numeric, 2), - 0 - ) AS today_avg_people_count - FROM admin.capture_logs - WHERE status = 'success' - AND measured_at >= date_trunc('day', now()) - AND measured_at < date_trunc('day', now()) + interval '1 day' - ) - INSERT INTO dashboard.current_status ( - id, - current_people_count, - updated_at, - today_max_people_count, - today_avg_people_count - ) - SELECT - 1, - %s, - %s, - today_max_people_count, - today_avg_people_count - FROM today_summary - ON CONFLICT (id) - DO UPDATE SET - current_people_count = EXCLUDED.current_people_count, - updated_at = EXCLUDED.updated_at, - today_max_people_count = EXCLUDED.today_max_people_count, - today_avg_people_count = EXCLUDED.today_avg_people_count - """, - (people_count, measured_at), - ) - - # 3. dashboard.today_hourly_stats 갱신 - cur.execute( - """ - WITH hourly AS ( - SELECT - current_date AS stat_date, - EXTRACT(HOUR FROM measured_at)::int AS hour, - ROUND(AVG(people_count)::numeric, 2) AS avg_people_count, - MAX(people_count) AS max_people_count, - COUNT(*) AS sample_count - FROM admin.capture_logs - WHERE status = 'success' - AND measured_at >= date_trunc('day', now()) - AND measured_at < date_trunc('day', now()) + interval '1 day' - GROUP BY stat_date, hour - ) - INSERT INTO dashboard.today_hourly_stats ( - stat_date, - hour, - avg_people_count, - max_people_count, - sample_count, - updated_at - ) - SELECT - stat_date, - hour, - avg_people_count, - max_people_count, - sample_count, - now() - FROM hourly - ON CONFLICT (stat_date, hour) - DO UPDATE SET - avg_people_count = EXCLUDED.avg_people_count, - max_people_count = EXCLUDED.max_people_count, - sample_count = EXCLUDED.sample_count, - updated_at = EXCLUDED.updated_at - """ - ) - - # 4. dashboard.daily_stats 갱신 - cur.execute( - """ - WITH base AS ( - SELECT - people_count, - measured_at, - EXTRACT(HOUR FROM measured_at)::int AS hour - FROM admin.capture_logs - WHERE status = 'success' - AND measured_at >= date_trunc('day', now()) - AND measured_at < date_trunc('day', now()) + interval '1 day' - ), - hourly AS ( - SELECT - hour, - ROUND(AVG(people_count)::numeric, 2) AS avg_people_count - FROM base - WHERE hour BETWEEN 9 AND 22 - GROUP BY hour - ), - summary AS ( - SELECT - current_date AS stat_date, - COALESCE( - ROUND(AVG(people_count)::numeric, 2), - 0 - ) AS avg_people_count, - COALESCE(MAX(people_count), 0) AS max_people_count, - COALESCE(MIN(people_count), 0) AS min_people_count, - COUNT(*) AS sample_count - FROM base - ), - peak AS ( - SELECT hour AS peak_hour - FROM hourly - ORDER BY avg_people_count DESC, hour ASC - LIMIT 1 - ), - quiet AS ( - SELECT hour AS quiet_hour - FROM hourly - ORDER BY avg_people_count ASC, hour ASC - LIMIT 1 - ) - INSERT INTO dashboard.daily_stats ( - stat_date, - avg_people_count, - max_people_count, - min_people_count, - peak_hour, - quiet_hour, - sample_count, - updated_at - ) - SELECT - summary.stat_date, - summary.avg_people_count, - summary.max_people_count, - summary.min_people_count, - peak.peak_hour, - quiet.quiet_hour, - summary.sample_count, - now() - FROM summary - LEFT JOIN peak ON true - LEFT JOIN quiet ON true - ON CONFLICT (stat_date) - DO UPDATE SET - avg_people_count = EXCLUDED.avg_people_count, - max_people_count = EXCLUDED.max_people_count, - min_people_count = EXCLUDED.min_people_count, - peak_hour = EXCLUDED.peak_hour, - quiet_hour = EXCLUDED.quiet_hour, - sample_count = EXCLUDED.sample_count, - updated_at = EXCLUDED.updated_at - """ - ) - - # 5. dashboard.weekday_hourly_stats 갱신 - cur.execute( - """ - WITH weekday_hourly AS ( - SELECT - EXTRACT(ISODOW FROM measured_at)::int AS weekday, - EXTRACT(HOUR FROM measured_at)::int AS hour, - ROUND(AVG(people_count)::numeric, 2) AS avg_people_count, - MAX(people_count) AS max_people_count, - COUNT(*) AS sample_count - FROM admin.capture_logs - WHERE status = 'success' - AND measured_at >= now() - interval '4 weeks' - AND EXTRACT(ISODOW FROM measured_at)::int BETWEEN 1 AND 5 - AND EXTRACT(HOUR FROM measured_at)::int BETWEEN 9 AND 22 - GROUP BY weekday, hour - ) - INSERT INTO dashboard.weekday_hourly_stats ( - weekday, - hour, - avg_people_count, - max_people_count, - sample_count, - updated_at - ) - SELECT - weekday, - hour, - avg_people_count, - max_people_count, - sample_count, - now() - FROM weekday_hourly - ON CONFLICT (weekday, hour) - DO UPDATE SET - avg_people_count = EXCLUDED.avg_people_count, - max_people_count = EXCLUDED.max_people_count, - sample_count = EXCLUDED.sample_count, - updated_at = EXCLUDED.updated_at - """ - ) - - -def get_today_stats_from_db(): - with get_db_connection() as conn: - with conn.cursor() as cur: - cur.execute( - """ - SELECT hours.hour, COALESCE(stats.avg_people_count, 0) AS count - FROM generate_series(9, 22) AS hours(hour) - LEFT JOIN dashboard.today_hourly_stats AS stats - ON stats.hour = hours.hour - AND stats.stat_date = current_date - ORDER BY hours.hour - """ - ) - rows = cur.fetchall() - - return [{"hour": row[0], "count": row[1]} for row in rows] - - -def get_weekly_stats_from_db(): - day_key_map = {1: "mon", 2: "tue", 3: "wed", 4: "thu", 5: "fri"} - - weekly_stats = { - "mon": [0] * 14, - "tue": [0] * 14, - "wed": [0] * 14, - "thu": [0] * 14, - "fri": [0] * 14, - } - - with get_db_connection() as conn: - with conn.cursor() as cur: - cur.execute( - """ - SELECT - weekday, - hour, - avg_people_count - FROM dashboard.weekday_hourly_stats - WHERE weekday BETWEEN 1 AND 5 - AND hour BETWEEN 9 AND 22 - ORDER BY weekday, hour - """ - ) - rows = cur.fetchall() - for weekday, hour, avg_people_count in rows: - day_key = day_key_map.get(weekday) - if not day_key: - continue - - index = hour - 9 - if 0 <= index < 14: - weekly_stats[day_key][index] = to_float(avg_people_count) - - return weekly_stats +app.include_router(public_router) +app.include_router(device_router) # --------------------------------------------- -SHOULD_CAPTURE = False - - -@app.post("/api/v2/test/trigger") -async def trigger_capture(): - """ - [Swagger UI용] 이 API를 호출하면 ESP32에게 사진을 찍으라는 명령을 내립니다. - """ - global SHOULD_CAPTURE - SHOULD_CAPTURE = True - return {"status": "success", "message": "Capture command issued to ESP32."} - - -@app.get("/api/v2/device/command") -async def get_device_command(): - """ - [ESP32용] ESP32가 주기적으로 이 API를 호출하여 명령이 있는지 확인합니다. - """ - global SHOULD_CAPTURE - if SHOULD_CAPTURE: - SHOULD_CAPTURE = False # 명령을 전달했으므로 다시 대기 상태로 변경 - return {"command": "capture"} - return {"command": "idle"} - - -# -------------------------------------------- - @app.get("/") async def root(): @@ -471,93 +49,3 @@ async def root(): 브라우저로 서버 주소에 접속하면 index.html 웹페이지를 띄워줍니다. """ return FileResponse("../frontend/index.html") - - -@app.get("/api/v2/count/current") -async def get_current_count(): - """ - [프론트엔드용] 현재 동아리방 인원수를 반환합니다. - """ - return get_current_status_from_db() - - -@app.get("/api/v2/stats/today") -async def get_today_stats(): - """ - [프론트엔드용] 오늘 시간대별 인원 통계를 반환합니다. - """ - return get_today_stats_from_db() - - -@app.get("/api/v2/stats/weekly") -async def get_weekly_stats(): - """ - [프론트엔드용] 평일 시간대별 평균 통계를 반환합니다. - """ - return get_weekly_stats_from_db() - - -# -------------------------------------------- - - -@app.post("/api/v2/device/upload") -async def upload_image(file: Annotated[UploadFile, File(...)]): - """ - [ESP32용] ESP32-CAM이 업로드한 이미지를 서버 메모리에서만 처리합니다. - - 처리 흐름: - 1. 업로드된 이미지 bytes 읽기 - 2. OpenCV로 JPEG 디코딩 - 3. YOLO로 사람 수 추론 - 4. 최신 인원 수 갱신 - 5. 이미지 파일은 저장하지 않고 함수 종료와 함께 폐기 - """ - measured_at = datetime.now() - - try: - # 업로드된 이미지 bytes를 메모리에서 읽음 - content = await file.read() - - # bytes -> numpy array -> OpenCV frame - nparr = np.frombuffer(content, np.uint8) - frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR) - - if frame is None: - error_message = "Failed to decode image." - print(f"[{measured_at}] {error_message}") - - save_failed_inference(error_message=error_message, trigger_type="scheduled") - - return {"status": "error", "message": error_message} - - inference_start = time.perf_counter() - people_count = infer_people_count(frame) - inference_time_ms = int((time.perf_counter() - inference_start) * 1000) - - save_successful_inference( - people_count=people_count, - trigger_type="scheduled", - inference_time_ms=inference_time_ms, - ) - - print( - f"[{measured_at}] Inference Result: " - f"{people_count} people detected. " - f"Inference Time: {inference_time_ms} ms" - ) - - return { - "status": "success", - "message": "Image processed successfully", - "detected_count": people_count, - } - except Exception as e: - error_message = str(e) - print(f"Error during upload/inference: {error_message}") - - try: - save_failed_inference(error_message=error_message, trigger_type="scheduled") - except Exception as db_error: - print(f"Failed to save error log to DB: {str(db_error)}") - - return {"status": "error", "message": error_message} diff --git a/backend/routers/__init__.py b/backend/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/routers/device.py b/backend/routers/device.py new file mode 100644 index 0000000..9598b7a --- /dev/null +++ b/backend/routers/device.py @@ -0,0 +1,105 @@ +from datetime import datetime +from typing import Annotated + +from fastapi import APIRouter, File, UploadFile + +from services.inference_service import decode_image, run_inference +from services.stats_service import ( + save_failed_inference, + save_successful_inference, +) + +router = APIRouter(prefix="/api/v2", tags=["device"]) + +_should_capture = False + + +@router.post("/test/trigger") +async def trigger_capture(): + """ + [Swagger UI용] ESP32에 사진 촬영 명령을 전달합니다. + """ + global _should_capture + + _should_capture = True + + return { + "status": "success", + "message": "Capture command issued to ESP32.", + } + + +@router.get("/device/command") +async def get_device_command(): + """ + [ESP32용] ESP32가 촬영 명령을 확인합니다. + """ + global _should_capture + + if _should_capture: + _should_capture = False + return {"command": "capture"} + + return {"command": "idle"} + + +@router.post("/device/upload") +async def upload_image(file: Annotated[UploadFile, File(...)]): + """ + [ESP32용] 업로드된 이미지를 메모리에서 처리하고 추론 결과를 저장합니다. + """ + measured_at = datetime.now() + + try: + content = await file.read() + frame = decode_image(content) + + if frame is None: + error_message = "Failed to decode image." + print(f"[{measured_at}] {error_message}") + + save_failed_inference( + error_message=error_message, + trigger_type="scheduled", + ) + + return { + "status": "error", + "message": error_message, + } + + people_count, inference_time_ms = run_inference(frame) + + save_successful_inference( + people_count=people_count, + trigger_type="scheduled", + inference_time_ms=inference_time_ms, + ) + + print( + f"[{measured_at}] Inference Result: " + f"{people_count} people detected. " + f"Inference Time: {inference_time_ms} ms" + ) + + return { + "status": "success", + "message": "Image processed successfully", + "detected_count": people_count, + } + except Exception as error: + error_message = str(error) + print(f"Error during upload/inference: {error_message}") + + try: + save_failed_inference( + error_message=error_message, + trigger_type="scheduled", + ) + except Exception as db_error: + print(f"Failed to save error log to DB: {db_error}") + + return { + "status": "error", + "message": error_message, + } diff --git a/backend/routers/public.py b/backend/routers/public.py new file mode 100644 index 0000000..a317c4b --- /dev/null +++ b/backend/routers/public.py @@ -0,0 +1,33 @@ +from fastapi import APIRouter + +from services.stats_service import ( + get_current_status_from_db, + get_today_stats_from_db, + get_weekly_stats_from_db, +) + +router = APIRouter(prefix="/api/v2", tags=["public"]) + + +@router.get("/count/current") +async def get_current_count(): + """ + [프론트엔드용] 현재 동아리방 인원수를 반환합니다. + """ + return get_current_status_from_db() + + +@router.get("/stats/today") +async def get_today_stats(): + """ + [프론트엔드용] 오늘 시간대별 인원 통계를 반환합니다. + """ + return get_today_stats_from_db() + + +@router.get("/stats/weekly") +async def get_weekly_stats(): + """ + [프론트엔드용] 평일 시간대별 평균 통계를 반환합니다. + """ + return get_weekly_stats_from_db() diff --git a/backend/services/__init__.py b/backend/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/services/inference_service.py b/backend/services/inference_service.py new file mode 100644 index 0000000..9222692 --- /dev/null +++ b/backend/services/inference_service.py @@ -0,0 +1,37 @@ +import time + +import cv2 +import numpy as np +from ultralytics import YOLO + +print("Loading YOLO model...") +_model = YOLO("yolov8n.pt") +print("YOLO model loaded successfully.") + + +def decode_image(content: bytes) -> np.ndarray | None: + """JPEG 이미지 바이트를 OpenCV frame으로 디코딩합니다.""" + image_buffer = np.frombuffer(content, dtype=np.uint8) + return cv2.imdecode(image_buffer, cv2.IMREAD_COLOR) + + +def infer_people_count(frame: np.ndarray) -> int: + """OpenCV frame에서 사람 수를 추론합니다.""" + results = _model.predict( + source=frame, + conf=0.3, + classes=[0], + imgsz=416, + verbose=False, + ) + + return int((results[0].boxes.cls == 0).sum().item()) + + +def run_inference(frame: np.ndarray) -> tuple[int, int]: + """사람 수를 추론하고 소요 시간을 밀리초 단위로 반환합니다.""" + inference_start = time.perf_counter() + people_count = infer_people_count(frame) + inference_time_ms = int((time.perf_counter() - inference_start) * 1000) + + return people_count, inference_time_ms diff --git a/backend/services/stats_service.py b/backend/services/stats_service.py new file mode 100644 index 0000000..d6f9a4b --- /dev/null +++ b/backend/services/stats_service.py @@ -0,0 +1,362 @@ +from decimal import Decimal + +from database import get_db_connection + + +def to_float(value): + if value is None: + return 0.0 + if isinstance(value, Decimal): + return float(value) + return value + + +def get_current_status_from_db(): + with get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT + current_people_count, + updated_at, + today_max_people_count, + today_avg_people_count + FROM dashboard.current_status + WHERE id = 1 + AND updated_at >= date_trunc('day', now()) + AND updated_at < date_trunc('day', now()) + interval '1 day' + """ + ) + row = cur.fetchone() + + if row is None: + return { + "count": 0, + "updated_at": None, + "today_max_count": 0, + "today_avg_count": 0.0, + } + + return { + "count": row[0], + "updated_at": row[1].isoformat() if row[1] else None, + "today_max_count": row[2], + "today_avg_count": to_float(row[3]), + } + + +def save_failed_inference( + error_message: str, + trigger_type: str = "scheduled", + inference_time_ms: int | None = None, +): + with get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO admin.capture_logs ( + people_count, + status, + trigger_type, + inference_time_ms, + error_message + ) + VALUES (%s, %s, %s, %s, %s) + """, + (None, "failed", trigger_type, inference_time_ms, error_message), + ) + + +def save_successful_inference( + people_count: int, + trigger_type: str = "scheduled", + inference_time_ms: int | None = None, +): + """ + YOLO 추론 성공 결과를 admin raw log와 dashboard summary tables에 반영합니다. + + 처리 순서: + 1. admin.capture_logs insert + 2. dashboard.current_status update + 3. dashboard.today_hourly_stats update + 4. dashboard.daily_stats update + 5. dashboard.weekday_hourly_stats update + """ + with get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO admin.capture_logs ( + people_count, + status, + trigger_type, + inference_time_ms + ) + VALUES (%s, %s, %s, %s) + RETURNING measured_at + """, + (people_count, "success", trigger_type, inference_time_ms), + ) + measured_at = cur.fetchone()[0] + + cur.execute( + """ + WITH today_summary AS ( + SELECT + COALESCE(MAX(people_count), 0) AS today_max_people_count, + COALESCE( + ROUND(AVG(people_count)::numeric, 2), + 0 + ) AS today_avg_people_count + FROM admin.capture_logs + WHERE status = 'success' + AND measured_at >= date_trunc('day', now()) + AND measured_at < date_trunc('day', now()) + interval '1 day' + ) + INSERT INTO dashboard.current_status ( + id, + current_people_count, + updated_at, + today_max_people_count, + today_avg_people_count + ) + SELECT + 1, + %s, + %s, + today_max_people_count, + today_avg_people_count + FROM today_summary + ON CONFLICT (id) + DO UPDATE SET + current_people_count = EXCLUDED.current_people_count, + updated_at = EXCLUDED.updated_at, + today_max_people_count = EXCLUDED.today_max_people_count, + today_avg_people_count = EXCLUDED.today_avg_people_count + """, + (people_count, measured_at), + ) + + cur.execute( + """ + WITH hourly AS ( + SELECT + current_date AS stat_date, + EXTRACT(HOUR FROM measured_at)::int AS hour, + ROUND(AVG(people_count)::numeric, 2) AS avg_people_count, + MAX(people_count) AS max_people_count, + COUNT(*) AS sample_count + FROM admin.capture_logs + WHERE status = 'success' + AND measured_at >= date_trunc('day', now()) + AND measured_at < date_trunc('day', now()) + interval '1 day' + GROUP BY stat_date, hour + ) + INSERT INTO dashboard.today_hourly_stats ( + stat_date, + hour, + avg_people_count, + max_people_count, + sample_count, + updated_at + ) + SELECT + stat_date, + hour, + avg_people_count, + max_people_count, + sample_count, + now() + FROM hourly + ON CONFLICT (stat_date, hour) + DO UPDATE SET + avg_people_count = EXCLUDED.avg_people_count, + max_people_count = EXCLUDED.max_people_count, + sample_count = EXCLUDED.sample_count, + updated_at = EXCLUDED.updated_at + """ + ) + + cur.execute( + """ + WITH base AS ( + SELECT + people_count, + measured_at, + EXTRACT(HOUR FROM measured_at)::int AS hour + FROM admin.capture_logs + WHERE status = 'success' + AND measured_at >= date_trunc('day', now()) + AND measured_at < date_trunc('day', now()) + interval '1 day' + ), + hourly AS ( + SELECT + hour, + ROUND(AVG(people_count)::numeric, 2) AS avg_people_count + FROM base + WHERE hour BETWEEN 9 AND 22 + GROUP BY hour + ), + summary AS ( + SELECT + current_date AS stat_date, + COALESCE( + ROUND(AVG(people_count)::numeric, 2), + 0 + ) AS avg_people_count, + COALESCE(MAX(people_count), 0) AS max_people_count, + COALESCE(MIN(people_count), 0) AS min_people_count, + COUNT(*) AS sample_count + FROM base + ), + peak AS ( + SELECT hour AS peak_hour + FROM hourly + ORDER BY avg_people_count DESC, hour ASC + LIMIT 1 + ), + quiet AS ( + SELECT hour AS quiet_hour + FROM hourly + ORDER BY avg_people_count ASC, hour ASC + LIMIT 1 + ) + INSERT INTO dashboard.daily_stats ( + stat_date, + avg_people_count, + max_people_count, + min_people_count, + peak_hour, + quiet_hour, + sample_count, + updated_at + ) + SELECT + summary.stat_date, + summary.avg_people_count, + summary.max_people_count, + summary.min_people_count, + peak.peak_hour, + quiet.quiet_hour, + summary.sample_count, + now() + FROM summary + LEFT JOIN peak ON true + LEFT JOIN quiet ON true + ON CONFLICT (stat_date) + DO UPDATE SET + avg_people_count = EXCLUDED.avg_people_count, + max_people_count = EXCLUDED.max_people_count, + min_people_count = EXCLUDED.min_people_count, + peak_hour = EXCLUDED.peak_hour, + quiet_hour = EXCLUDED.quiet_hour, + sample_count = EXCLUDED.sample_count, + updated_at = EXCLUDED.updated_at + """ + ) + + cur.execute( + """ + WITH weekday_hourly AS ( + SELECT + EXTRACT(ISODOW FROM measured_at)::int AS weekday, + EXTRACT(HOUR FROM measured_at)::int AS hour, + ROUND(AVG(people_count)::numeric, 2) AS avg_people_count, + MAX(people_count) AS max_people_count, + COUNT(*) AS sample_count + FROM admin.capture_logs + WHERE status = 'success' + AND measured_at >= now() - interval '4 weeks' + AND EXTRACT(ISODOW FROM measured_at)::int BETWEEN 1 AND 5 + AND EXTRACT(HOUR FROM measured_at)::int BETWEEN 9 AND 22 + GROUP BY weekday, hour + ) + INSERT INTO dashboard.weekday_hourly_stats ( + weekday, + hour, + avg_people_count, + max_people_count, + sample_count, + updated_at + ) + SELECT + weekday, + hour, + avg_people_count, + max_people_count, + sample_count, + now() + FROM weekday_hourly + ON CONFLICT (weekday, hour) + DO UPDATE SET + avg_people_count = EXCLUDED.avg_people_count, + max_people_count = EXCLUDED.max_people_count, + sample_count = EXCLUDED.sample_count, + updated_at = EXCLUDED.updated_at + """ + ) + + +def get_today_stats_from_db(): + with get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT hours.hour, COALESCE(stats.avg_people_count, 0) AS count + FROM generate_series(9, 22) AS hours(hour) + LEFT JOIN dashboard.today_hourly_stats AS stats + ON stats.hour = hours.hour + AND stats.stat_date = current_date + ORDER BY hours.hour + """ + ) + rows = cur.fetchall() + + return [{"hour": row[0], "count": row[1]} for row in rows] + + +def get_weekly_stats_from_db(): + day_key_map = { + 1: "mon", + 2: "tue", + 3: "wed", + 4: "thu", + 5: "fri", + } + + weekly_stats = { + "mon": [0] * 14, + "tue": [0] * 14, + "wed": [0] * 14, + "thu": [0] * 14, + "fri": [0] * 14, + } + + with get_db_connection() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT + weekday, + hour, + avg_people_count + FROM dashboard.weekday_hourly_stats + WHERE weekday BETWEEN 1 AND 5 + AND hour BETWEEN 9 AND 22 + ORDER BY weekday, hour + """ + ) + rows = cur.fetchall() + + for weekday, hour, avg_people_count in rows: + day_key = day_key_map.get(weekday) + + if not day_key: + continue + + index = hour - 9 + + if 0 <= index < 14: + weekly_stats[day_key][index] = to_float(avg_people_count) + + return weekly_stats