/
volkovss
/
eye_tracker
Обзор
Документация
Войти
/
volkovss
/
eye_tracker
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
api/server.py
682 строки
22 KB
volkovss
Initial commit: Eye Tracker v1.0
03 июн 2026, 13:18
03 июн 2026, 13:18
f9d57a2
Код
Авторство
О чём код?
import hashlib import io import json import csv import secrets from datetime import datetime from typing import Optional from fastapi import FastAPI, Request, Form, Response, HTTPException from fastapi.responses import ( HTMLResponse, RedirectResponse, JSONResponse, StreamingResponse, FileResponse, ) from fastapi.templating import Jinja2Templates from fastapi.staticfiles import StaticFiles import os, sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from database.models import ( init_db, get_db_session, User, Device, Session as DBSession, GazeData, CalibrationData, HardwareConfig, SoftwareConfig, EnvironmentParams, AggregatedStats, Log, ) from analytics.session_analyzer import SessionAnalyzer # приложение app = FastAPI(title="Eye Tracker Web") BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) TMPL_DIR = os.path.join(BASE_DIR, "web", "templates") STAT_DIR = os.path.join(BASE_DIR, "web", "static") os.makedirs(STAT_DIR, exist_ok=True) os.makedirs(os.path.join(STAT_DIR, "heatmaps"), exist_ok=True) templates = Jinja2Templates(directory=TMPL_DIR) app.mount("/static", StaticFiles(directory=STAT_DIR), name="static") engine = init_db() analyzer = SessionAnalyzer(engine) # простая сессионная аутентификация _sessions: dict = {} # token → user_info dict def _make_token(user: User) -> str: token = secrets.token_urlsafe(32) _sessions[token] = { "user_id": user.user_id, "name": user.name, "email": user.email, "role": user.role, } return token def _current_user(request: Request) -> Optional[dict]: token = request.cookies.get("et_token") return _sessions.get(token) if token else None def _require_user(request: Request) -> dict: u = _current_user(request) if not u: raise HTTPException(status_code=302, headers={"Location": "/login"}) return u def _require_admin(request: Request) -> dict: u = _require_user(request) if u["role"] != "admin": raise HTTPException(status_code=403, detail="Требуются права администратора") return u def _tmpl(name: str, request: Request, extra: dict = None, user: dict = None): ctx = {"user": user or _current_user(request)} if extra: ctx.update(extra) return templates.TemplateResponse(request=request, name=name, context=ctx) # # AUTH # @app.get("/login", response_class=HTMLResponse) def login_page(request: Request): if _current_user(request): return RedirectResponse("/") return _tmpl("login.html", request) @app.post("/login") def login_post(request: Request, email: str = Form(...), password: str = Form(...)): db = get_db_session(engine) user = db.query(User).filter_by(email=email, is_active=True).first() ph = hashlib.sha256(password.encode()).hexdigest() db.close() if not user or user.password_hash != ph: return _tmpl("login.html", request, {"error": "Неверный email или пароль"}) token = _make_token(user) response = RedirectResponse("/", status_code=302) response.set_cookie("et_token", token, httponly=True, max_age=86400) return response @app.get("/logout") def logout(request: Request): token = request.cookies.get("et_token") _sessions.pop(token, None) resp = RedirectResponse("/login", status_code=302) resp.delete_cookie("et_token") return resp # # DASHBOARD # @app.get("/", response_class=HTMLResponse) def dashboard(request: Request): user = _require_user(request) db = get_db_session(engine) total_sessions = db.query(DBSession).count() total_users = db.query(User).count() total_points = db.query(GazeData).count() recent_sessions = ( db.query(DBSession) .order_by(DBSession.start_ts.desc()) .limit(8).all() ) db.close() return _tmpl("dashboard.html", request, { "total_sessions": total_sessions, "total_users": total_users, "total_points": total_points, "recent_sessions": recent_sessions, }, user=user) # # SESSIONS # @app.get("/sessions", response_class=HTMLResponse) def sessions_list(request: Request, page: int = 1, user_id: int = 0): user = _require_user(request) db = get_db_session(engine) PER = 15 q = db.query(DBSession) if user_id: q = q.filter_by(user_id=user_id) total = q.count() sessions = (q.order_by(DBSession.start_ts.desc()) .offset((page - 1) * PER).limit(PER).all()) users = db.query(User).all() gaze_counts = { s.session_id: db.query(GazeData).filter_by(session_id=s.session_id).count() for s in sessions } db.close() return _tmpl("sessions.html", request, { "sessions": sessions, "users": users, "page": page, "total": total, "per": PER, "filter_user_id": user_id, "gaze_counts": gaze_counts, }, user=user) @app.get("/sessions/new", response_class=HTMLResponse) def session_new_page(request: Request): user = _require_user(request) db = get_db_session(engine) users = db.query(User).filter_by(is_active=True).all() devices = db.query(Device).all() db.close() return _tmpl("session_new.html", request, {"users": users, "devices": devices}, user=user) @app.get("/sessions/{sid}", response_class=HTMLResponse) def session_detail(sid: int, request: Request): user = _require_user(request) db = get_db_session(engine) sess = db.query(DBSession).filter_by(session_id=sid).first() if not sess: raise HTTPException(404, "Сессия не найдена") agg = db.query(AggregatedStats).filter_by(session_id=sid).first() hw = db.query(HardwareConfig).filter_by(session_id=sid).first() sw = db.query(SoftwareConfig).filter_by(session_id=sid).first() env = db.query(EnvironmentParams).filter_by(session_id=sid).first() cal = (db.query(CalibrationData).filter_by(session_id=sid) .order_by(CalibrationData.cal_id.desc()).first()) db.close() return _tmpl("session_detail.html", request, { "sess": sess, "agg": agg, "hw": hw, "sw": sw, "env": env, "cal": cal, }, user=user) @app.post("/sessions/{sid}/analyze") def analyze_session(sid: int, request: Request): _require_user(request) result = analyzer.analyze(sid) return JSONResponse(result) @app.get("/sessions/{sid}/heatmap.png") def heatmap_img(sid: int, request: Request): _require_user(request) path = os.path.join(analyzer.HEATMAP_DIR, f"s{sid}_heatmap.png") if not os.path.exists(path): raise HTTPException(404, "Тепловая карта не найдена. Запустите анализ.") return FileResponse(path, media_type="image/png") @app.get("/sessions/{sid}/scanpath.png") def scanpath_img(sid: int, request: Request): _require_user(request) path = os.path.join(analyzer.HEATMAP_DIR, f"s{sid}_scanpath.png") if not os.path.exists(path): raise HTTPException(404, "Scanpath не найден. Запустите анализ.") return FileResponse(path, media_type="image/png") @app.get("/sessions/{sid}/gaze-json") def gaze_json(sid: int, request: Request, limit: int = 500): _require_user(request) db = get_db_session(engine) rows = (db.query(GazeData).filter_by(session_id=sid) .order_by(GazeData.frame_index) .limit(limit).all()) db.close() data = [{"t": r.timestamp, "x": r.x_px, "y": r.y_px, "conf": r.confidence, "fi": r.frame_index} for r in rows] return JSONResponse(data) @app.post("/sessions/{sid}/delete") def delete_session(sid: int, request: Request): _require_admin(request) db = get_db_session(engine) db.query(GazeData).filter_by(session_id=sid).delete() db.query(AggregatedStats).filter_by(session_id=sid).delete() db.query(Log).filter_by(session_id=sid).delete() db.query(HardwareConfig).filter_by(session_id=sid).delete() db.query(SoftwareConfig).filter_by(session_id=sid).delete() db.query(EnvironmentParams).filter_by(session_id=sid).delete() db.query(CalibrationData).filter_by(session_id=sid).delete() db.query(DBSession).filter_by(session_id=sid).delete() db.commit() db.close() return RedirectResponse("/sessions", status_code=302) # # REPORTS # @app.get("/reports", response_class=HTMLResponse) def reports(request: Request): user = _require_user(request) db = get_db_session(engine) aggs = db.query(AggregatedStats).all() per_device = {} for a in aggs: sess = db.query(DBSession).filter_by(session_id=a.session_id).first() if sess and sess.device_id: dev = db.query(Device).filter_by(device_id=sess.device_id).first() key = dev.name if dev else f"Device {sess.device_id}" if key not in per_device: per_device[key] = {"sessions": 0, "avg_quality": 0, "total_fixations": 0} per_device[key]["sessions"] += 1 per_device[key]["avg_quality"] += (a.data_quality_pct or 0) per_device[key]["total_fixations"]+= (a.total_fixations or 0) for k in per_device: n = per_device[k]["sessions"] per_device[k]["avg_quality"] = round(per_device[k]["avg_quality"] / n, 1) db.close() return _tmpl("reports.html", request, {"aggs": aggs, "per_device": per_device}, user=user) # # FIXATION TIMELINE # @app.get("/sessions/{sid}/timeline", response_class=HTMLResponse) def fixation_timeline(sid: int, request: Request): user = _require_user(request) db = get_db_session(engine) sess = db.query(DBSession).filter_by(session_id=sid).first() db.close() return _tmpl("fixation_timeline.html", request, {"sess": sess, "sid": sid}, user=user) # # COMPARE # @app.get("/compare", response_class=HTMLResponse) def compare(request: Request): user = _require_user(request) db = get_db_session(engine) sessions = db.query(DBSession).order_by(DBSession.start_ts.desc()).all() db.close() return _tmpl("compare.html", request, {"sessions": sessions}, user=user) @app.get("/compare/data") def compare_data(request: Request, a: int = 0, b: int = 0): _require_user(request) result = {} for sid in [a, b]: if sid: s = analyzer.get_summary(sid) result[str(sid)] = s return JSONResponse(result) # # AOI ANALYSIS # @app.get("/aoi", response_class=HTMLResponse) def aoi_page(request: Request, sid: int = 0): user = _require_user(request) db = get_db_session(engine) sessions = db.query(DBSession).order_by(DBSession.start_ts.desc()).all() db.close() return _tmpl("aoi.html", request, {"sessions": sessions, "sel_sid": sid}, user=user) @app.post("/aoi/compute") async def aoi_compute(request: Request): _require_user(request) body = await request.json() sid = int(body.get("session_id", 0)) aoi_defs= body.get("aois", []) from analytics.aoi import AOI, analyze_aois db = get_db_session(engine) rows = db.query(GazeData).filter_by(session_id=sid).all() db.close() points = [{"x_px": r.x_px, "y_px": r.y_px, "confidence": r.confidence} for r in rows] aois = [AOI(a["name"], a["x1"], a["y1"], a["x2"], a["y2"]) for a in aoi_defs] metrics = analyze_aois(points, aois) return JSONResponse([ {"name": m.name, "total_time_ms": m.total_time_ms, "fixation_count": m.fixation_count, "avg_fixation_ms": m.avg_fixation_ms, "entry_count": m.entry_count, "pct_total_time": m.pct_total_time} for m in metrics ]) # # EXPORT # @app.get("/export", response_class=HTMLResponse) def export_page(request: Request): user = _require_user(request) db = get_db_session(engine) sessions = db.query(DBSession).order_by(DBSession.start_ts.desc()).all() db.close() return _tmpl("export.html", request, {"sessions": sessions}, user=user) @app.get("/export/{sid}/csv") def export_csv(sid: int, request: Request): _require_user(request) db = get_db_session(engine) rows = (db.query(GazeData).filter_by(session_id=sid) .order_by(GazeData.frame_index).all()) db.close() buf = io.StringIO() w = csv.writer(buf) w.writerow(["gaze_id","session_id","timestamp","frame_index", "x_px","y_px","x_norm","y_norm","confidence", "left_pupil_x","left_pupil_y","right_pupil_x","right_pupil_y"]) for r in rows: w.writerow([r.gaze_id, r.session_id, r.timestamp, r.frame_index, r.x_px, r.y_px, r.x_norm, r.y_norm, r.confidence, r.left_pupil_x, r.left_pupil_y, r.right_pupil_x, r.right_pupil_y]) buf.seek(0) return StreamingResponse( io.BytesIO(buf.getvalue().encode()), media_type="text/csv", headers={"Content-Disposition": f"attachment; filename=session_{sid}.csv"}, ) @app.get("/export/{sid}/json") def export_json_file(sid: int, request: Request): _require_user(request) db = get_db_session(engine) rows = (db.query(GazeData).filter_by(session_id=sid) .order_by(GazeData.frame_index).all()) sess = db.query(DBSession).filter_by(session_id=sid).first() db.close() payload = { "session_id": sid, "task_name": sess.task_name if sess else "", "exported_at":datetime.utcnow().isoformat(), "gaze_data": [ {"timestamp": r.timestamp, "frame_index": r.frame_index, "x_px": r.x_px, "y_px": r.y_px, "x_norm": r.x_norm, "y_norm": r.y_norm, "confidence": r.confidence} for r in rows ], } raw = json.dumps(payload, ensure_ascii=False, indent=2).encode() return StreamingResponse( io.BytesIO(raw), media_type="application/json", headers={"Content-Disposition": f"attachment; filename=session_{sid}.json"}, ) # # CALIBRATION HISTORY # @app.get("/calibrations", response_class=HTMLResponse) def calibrations(request: Request): user = _require_user(request) db = get_db_session(engine) cals = db.query(CalibrationData).order_by(CalibrationData.cal_id.desc()).all() db.close() return _tmpl("calibration_history.html", request, {"cals": cals}, user=user) # # LOGS # @app.get("/logs", response_class=HTMLResponse) def logs(request: Request, level: str = ""): user = _require_user(request) db = get_db_session(engine) q = db.query(Log) if level: q = q.filter_by(level=level) logs_data = q.order_by(Log.ts.desc()).limit(300).all() db.close() return _tmpl("logs.html", request, {"logs": logs_data, "level_filter": level}, user=user) # # SUBJECTS # @app.get("/subjects", response_class=HTMLResponse) def subjects(request: Request): user = _require_user(request) db = get_db_session(engine) subs = db.query(User).filter_by(role="subject").all() db.close() return _tmpl("subjects.html", request, {"subjects": subs}, user=user) # # ADMIN — ПОЛЬЗОВАТЕЛИ # @app.get("/admin/users", response_class=HTMLResponse) def admin_users(request: Request): _ = _require_admin(request) user = _current_user(request) db = get_db_session(engine) users = db.query(User).order_by(User.user_id).all() db.close() return _tmpl("admin_users.html", request, {"users": users}, user=user) @app.get("/admin/users/new", response_class=HTMLResponse) def user_new_page(request: Request): _ = _require_admin(request) return _tmpl("user_form.html", request, {"edit": None}) @app.post("/admin/users/new") def user_new_post(request: Request, name: str = Form(...), email: str = Form(...), password: str = Form(...), role: str = Form(...), consent: str = Form(default="")): _ = _require_admin(request) db = get_db_session(engine) db.add(User( name=name, email=email, password_hash=hashlib.sha256(password.encode()).hexdigest(), role=role, consent_flag=bool(consent), )) db.commit() db.close() return RedirectResponse("/admin/users", status_code=302) @app.get("/admin/users/{uid}/edit", response_class=HTMLResponse) def user_edit_page(uid: int, request: Request): _ = _require_admin(request) db = get_db_session(engine) user_obj = db.query(User).filter_by(user_id=uid).first() db.close() return _tmpl("user_form.html", request, {"edit": user_obj}) @app.post("/admin/users/{uid}/edit") def user_edit_post(uid: int, request: Request, name: str = Form(...), email: str = Form(...), role: str = Form(...), consent: str = Form(default=""), password:str = Form(default="")): _ = _require_admin(request) db = get_db_session(engine) u = db.query(User).filter_by(user_id=uid).first() if u: u.name = name u.email = email u.role = role u.consent_flag = bool(consent) if password.strip(): u.password_hash = hashlib.sha256(password.encode()).hexdigest() db.commit() db.close() return RedirectResponse("/admin/users", status_code=302) @app.post("/admin/users/{uid}/delete") def user_delete(uid: int, request: Request): _ = _require_admin(request) db = get_db_session(engine) db.query(User).filter_by(user_id=uid).delete() db.commit() db.close() return RedirectResponse("/admin/users", status_code=302) # # ADMIN — УСТРОЙСТВА # @app.get("/admin/devices", response_class=HTMLResponse) def admin_devices(request: Request): _ = _require_admin(request) user = _current_user(request) db = get_db_session(engine) devs = db.query(Device).all() db.close() return _tmpl("admin_devices.html", request, {"devices": devs}, user=user) @app.get("/admin/devices/new", response_class=HTMLResponse) def device_new_page(request: Request): _ = _require_admin(request) return _tmpl("device_form.html", request, {"edit": None}) @app.post("/admin/devices/new") def device_new_post(request: Request, name: str = Form(...), manufacturer: str = Form(default=""), model: str = Form(default=""), resolution: str = Form(default="640x480"), fps: int = Form(default=60), cam_left: int = Form(default=0), cam_right: int = Form(default=1), notes: str = Form(default="")): _ = _require_admin(request) db = get_db_session(engine) db.add(Device(name=name, manufacturer=manufacturer, model=model, camera_resolution=resolution, camera_fps=fps, cam_left_index=cam_left, cam_right_index=cam_right, ir_enabled=True, notes=notes)) db.commit() db.close() return RedirectResponse("/admin/devices", status_code=302) @app.get("/admin/devices/{did}/edit", response_class=HTMLResponse) def device_edit_page(did: int, request: Request): _ = _require_admin(request) db = get_db_session(engine) dev = db.query(Device).filter_by(device_id=did).first() db.close() return _tmpl("device_form.html", request, {"edit": dev}) @app.post("/admin/devices/{did}/edit") def device_edit_post(did: int, request: Request, name: str = Form(...), manufacturer: str = Form(default=""), model: str = Form(default=""), resolution: str = Form(default="640x480"), fps: int = Form(default=60), cam_left: int = Form(default=0), cam_right: int = Form(default=1), notes: str = Form(default="")): _ = _require_admin(request) db = get_db_session(engine) dev = db.query(Device).filter_by(device_id=did).first() if dev: dev.name=name; dev.manufacturer=manufacturer; dev.model=model dev.camera_resolution=resolution; dev.camera_fps=fps dev.cam_left_index=cam_left; dev.cam_right_index=cam_right; dev.notes=notes db.commit() db.close() return RedirectResponse("/admin/devices", status_code=302) @app.post("/admin/devices/{did}/delete") def device_delete(did: int, request: Request): _ = _require_admin(request) db = get_db_session(engine) db.query(Device).filter_by(device_id=did).delete() db.commit() db.close() return RedirectResponse("/admin/devices", status_code=302) # # PROFILE / SETTINGS / HELP / ABOUT # @app.get("/profile", response_class=HTMLResponse) def profile(request: Request): user = _require_user(request) db = get_db_session(engine) u = db.query(User).filter_by(user_id=user["user_id"]).first() db.close() return _tmpl("profile.html", request, {"u": u}, user=user) @app.post("/profile") def profile_post(request: Request, name: str = Form(...), password: str = Form(default="")): user = _require_user(request) db = get_db_session(engine) u = db.query(User).filter_by(user_id=user["user_id"]).first() if u: u.name = name if password.strip(): u.password_hash = hashlib.sha256(password.encode()).hexdigest() db.commit() _sessions[request.cookies.get("et_token", "")]["name"] = name db.close() return RedirectResponse("/profile", status_code=302) @app.get("/settings", response_class=HTMLResponse) def settings(request: Request): user = _require_user(request) return _tmpl("settings.html", request, {}, user=user) @app.get("/help", response_class=HTMLResponse) def help_page(request: Request): user = _require_user(request) return _tmpl("help.html", request, {}, user=user) @app.get("/about", response_class=HTMLResponse) def about(request: Request): user = _require_user(request) return _tmpl("about.html", request, {}, user=user)