Shoraky commited on
Commit
f514c45
·
1 Parent(s): 88fc756

Remove completed playback backfill tooling

Browse files
Files changed (5) hide show
  1. Dockerfile +1 -1
  2. README.md +0 -2
  3. app.py +1 -53
  4. backfill_playback_times.py +0 -100
  5. src/analysis.ts +0 -29
Dockerfile CHANGED
@@ -17,7 +17,7 @@ COPY requirements.txt ./
17
  RUN pip install --no-cache-dir -r requirements.txt
18
  COPY --from=node-build --chown=appuser:appuser /app/dist ./dist
19
  COPY --from=node-build --chown=appuser:appuser /app/node_modules ./node_modules
20
- COPY --chown=appuser:appuser package.json app.py start.py storage_listing.py playback_timing.py backfill_playback_times.py ./
21
  USER appuser
22
  EXPOSE 7860
23
  CMD ["python", "start.py"]
 
17
  RUN pip install --no-cache-dir -r requirements.txt
18
  COPY --from=node-build --chown=appuser:appuser /app/dist ./dist
19
  COPY --from=node-build --chown=appuser:appuser /app/node_modules ./node_modules
20
+ COPY --chown=appuser:appuser package.json app.py start.py storage_listing.py playback_timing.py ./
21
  USER appuser
22
  EXPOSE 7860
23
  CMD ["python", "start.py"]
README.md CHANGED
@@ -108,8 +108,6 @@ Brevo credentials stay in Supabase SMTP settings and are not added to the browse
108
 
109
  Run `Database/add_action_playback_times.sql` in the Supabase SQL Editor before relying on saved action playback times. New completed actions store the presentation timestamps of their first frame and the frame after their last frame from the finalized camera video. Older actions remain playable using the FPS calculation until backfilled.
110
 
111
- To fill older completed sessions, run `python backfill_playback_times.py` from this directory with `SUPABASE_URL`, `SUPABASE_SERVICE_ROLE_KEY`, `STORAGE_REPO_ID`, and `HF_TOKEN` set in `Backend/.env` or the environment. Add `--session-id UUID` to process one session. The script downloads one camera video per session and updates only actions whose playback times are missing, so the first run can take time for large videos. Re-running it is safe.
112
-
113
  ## Routes
114
 
115
  - `POST /api/auth/signup` — `{ fname, lname, email, password, academy? }`; sends Supabase's confirmation email when confirmation is enabled.
 
108
 
109
  Run `Database/add_action_playback_times.sql` in the Supabase SQL Editor before relying on saved action playback times. New completed actions store the presentation timestamps of their first frame and the frame after their last frame from the finalized camera video. Older actions remain playable using the FPS calculation until backfilled.
110
 
 
 
111
  ## Routes
112
 
113
  - `POST /api/auth/signup` — `{ fname, lname, email, password, academy? }`; sends Supabase's confirmation email when confirmation is enabled.
app.py CHANGED
@@ -17,7 +17,7 @@ from datetime import datetime, timezone
17
  from functools import lru_cache
18
  from pathlib import Path
19
  from urllib.request import urlretrieve
20
- from uuid import UUID, uuid4, uuid5
21
  from zipfile import ZipFile
22
 
23
  import httpx
@@ -37,8 +37,6 @@ load_dotenv(Path(__file__).with_name(".env"))
37
  jobs = {}
38
  executor = ThreadPoolExecutor(max_workers=1)
39
  logger = logging.getLogger("sporalize")
40
- playback_backfill_lock = threading.Lock()
41
- playback_backfill_job = None
42
  POSE_MODEL_FILE = "Weights/rtmpose-x-halpe26.onnx"
43
  POSE_MODEL_REPO_ID = ""
44
  POSE_MODEL_HUB_FILE = "rtmpose-x-halpe26.onnx"
@@ -250,28 +248,6 @@ async def lifespan(app):
250
  app = FastAPI(title="Sporalize AI", lifespan=lifespan)
251
 
252
 
253
- def run_playback_backfill_job(job_id):
254
- global playback_backfill_job
255
- from backfill_playback_times import run_backfill
256
-
257
- def update(progress):
258
- global playback_backfill_job
259
- with playback_backfill_lock:
260
- if playback_backfill_job and playback_backfill_job.get("job_id") == job_id:
261
- playback_backfill_job.update(progress)
262
-
263
- try:
264
- result = run_backfill(progress=update)
265
- with playback_backfill_lock:
266
- if playback_backfill_job and playback_backfill_job.get("job_id") == job_id:
267
- playback_backfill_job.update(result, status="completed")
268
- except Exception as error:
269
- logger.exception("Server-side action playback timestamp backfill failed")
270
- with playback_backfill_lock:
271
- if playback_backfill_job and playback_backfill_job.get("job_id") == job_id:
272
- playback_backfill_job.update(status="failed", error=str(error)[:500])
273
-
274
-
275
  @app.exception_handler(httpx.HTTPStatusError)
276
  async def backend_error(request: Request, error: httpx.HTTPStatusError):
277
  status = error.response.status_code
@@ -588,34 +564,6 @@ def health():
588
  return {"ok": True}
589
 
590
 
591
- @app.post("/internal/playback-backfill")
592
- def start_playback_backfill(request: Request, background_tasks: BackgroundTasks):
593
- global playback_backfill_job
594
- if not request.state.is_admin:
595
- raise HTTPException(403, "Administrator access is required")
596
- with playback_backfill_lock:
597
- if playback_backfill_job and playback_backfill_job.get("status") == "running":
598
- return {"job_id": playback_backfill_job["job_id"], "status": "running"}
599
- job_id = str(uuid4())
600
- playback_backfill_job = {
601
- "job_id": job_id, "status": "running", "stage": "starting",
602
- "sessions_total": 0, "sessions_completed": 0, "actions_total": 0,
603
- "actions_updated": 0, "errors": [],
604
- }
605
- background_tasks.add_task(run_playback_backfill_job, job_id)
606
- return {"job_id": job_id, "status": "running"}
607
-
608
-
609
- @app.get("/internal/playback-backfill/{job_id}")
610
- def playback_backfill_status(job_id: UUID, request: Request):
611
- if not request.state.is_admin:
612
- raise HTTPException(403, "Administrator access is required")
613
- with playback_backfill_lock:
614
- if not playback_backfill_job or playback_backfill_job.get("job_id") != str(job_id):
615
- raise HTTPException(404, "Backfill job not found")
616
- return dict(playback_backfill_job)
617
-
618
-
619
  @app.post("/api/drive/probe")
620
  def probe_drive(payload: dict):
621
  links = payload.get("links") or ([payload["link"]] if payload.get("link") else [])
 
17
  from functools import lru_cache
18
  from pathlib import Path
19
  from urllib.request import urlretrieve
20
+ from uuid import UUID, uuid5
21
  from zipfile import ZipFile
22
 
23
  import httpx
 
37
  jobs = {}
38
  executor = ThreadPoolExecutor(max_workers=1)
39
  logger = logging.getLogger("sporalize")
 
 
40
  POSE_MODEL_FILE = "Weights/rtmpose-x-halpe26.onnx"
41
  POSE_MODEL_REPO_ID = ""
42
  POSE_MODEL_HUB_FILE = "rtmpose-x-halpe26.onnx"
 
248
  app = FastAPI(title="Sporalize AI", lifespan=lifespan)
249
 
250
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
251
  @app.exception_handler(httpx.HTTPStatusError)
252
  async def backend_error(request: Request, error: httpx.HTTPStatusError):
253
  status = error.response.status_code
 
564
  return {"ok": True}
565
 
566
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
567
  @app.post("/api/drive/probe")
568
  def probe_drive(payload: dict):
569
  links = payload.get("links") or ([payload["link"]] if payload.get("link") else [])
backfill_playback_times.py DELETED
@@ -1,100 +0,0 @@
1
- """Fill action playback times from the stored final MP4 videos.
2
-
3
- Run after Database/add_action_playback_times.sql:
4
- python backfill_playback_times.py [--session-id UUID]
5
- """
6
-
7
- import argparse
8
- import os
9
- from collections import defaultdict
10
- from pathlib import Path
11
- from tempfile import TemporaryDirectory
12
-
13
- import httpx
14
- from dotenv import load_dotenv
15
- from huggingface_hub import HfApi, hf_hub_download
16
-
17
- from playback_timing import action_times, frame_times
18
- from storage_listing import session_files
19
-
20
-
21
- def run_backfill(session_id=None, progress=None):
22
- url = os.environ["SUPABASE_URL"].rstrip("/") + "/rest/v1"
23
- key = os.environ["SUPABASE_SERVICE_ROLE_KEY"]
24
- repo = os.environ["STORAGE_REPO_ID"]
25
- token = os.environ["HF_TOKEN"]
26
- hub = HfApi(token=token)
27
- headers = {"apikey": key, "Authorization": f"Bearer {key}"}
28
- actions = []
29
- with httpx.Client(headers=headers, timeout=60) as client:
30
- offset = 0
31
- while True:
32
- params = {"select": "action_id,session_id,start_frame,end_frame", "start_time_seconds": "is.null",
33
- "order": "action_id", "limit": "1000", "offset": str(offset)}
34
- if session_id:
35
- params["session_id"] = f"eq.{session_id}"
36
- response = client.get(f"{url}/actions", params=params)
37
- response.raise_for_status()
38
- page = response.json()
39
- actions.extend(page)
40
- if len(page) < 1000:
41
- break
42
- offset += len(page)
43
-
44
- grouped = defaultdict(list)
45
- for action in actions:
46
- grouped[action["session_id"]].append(action)
47
- saved = 0
48
- errors = []
49
- sessions = list(grouped.items())
50
- if progress:
51
- progress({"stage": "processing", "actions_total": len(actions), "sessions_total": len(sessions),
52
- "sessions_completed": 0, "actions_updated": 0, "errors": []})
53
- for index, (current_session_id, session_actions) in enumerate(sessions, start=1):
54
- try:
55
- response = client.get(f"{url}/sessions", params={"select": "path,status", "session_id": f"eq.{current_session_id}"})
56
- response.raise_for_status()
57
- session_rows = response.json()
58
- if len(session_rows) != 1 or session_rows[0]["status"] != "completed" or session_rows[0]["path"] != f"sessions/{current_session_id}":
59
- raise RuntimeError("Completed session storage unavailable")
60
- video_dir = f"sessions/{current_session_id}/videos"
61
- files = sorted(file for file in session_files(hub, repo, video_dir)
62
- if Path(file).name.startswith("camera_") and Path(file).suffix.lower() == ".mp4")
63
- if not files:
64
- raise RuntimeError("No MP4 video found")
65
- with TemporaryDirectory(prefix="playback-backfill-") as cache_dir:
66
- video = hf_hub_download(repo_id=repo, repo_type="dataset", filename=files[0], token=token,
67
- cache_dir=cache_dir)
68
- boundaries = frame_times(video)
69
- session_saved = 0
70
- for action in session_actions:
71
- times = action_times(boundaries, action["start_frame"], action["end_frame"])
72
- response = client.patch(f"{url}/actions", params={"action_id": f'eq.{action["action_id"]}',
73
- "start_time_seconds": "is.null"}, json=times)
74
- response.raise_for_status()
75
- session_saved += 1
76
- saved += session_saved
77
- print(f"Playback timestamp backfill: session {index}/{len(grouped)} {current_session_id} saved={session_saved}", flush=True)
78
- except Exception as error:
79
- errors.append({"session_id": current_session_id, "error": str(error)[:500]})
80
- print(f"Playback timestamp backfill: session {index}/{len(grouped)} {current_session_id} failed={error}", flush=True)
81
- if progress:
82
- progress({"stage": "processing", "actions_total": len(actions), "sessions_total": len(grouped),
83
- "sessions_completed": index, "actions_updated": saved, "errors": errors.copy(),
84
- "last_session_id": current_session_id})
85
- print(f"Playback timestamp backfill complete: {saved}/{len(actions)} action rows updated", flush=True)
86
- return {"actions_total": len(actions), "actions_updated": saved, "sessions_total": len(grouped),
87
- "sessions_completed": len(grouped), "errors": errors}
88
-
89
-
90
- def main():
91
- load_dotenv(Path(__file__).with_name(".env"))
92
- parser = argparse.ArgumentParser(description=__doc__)
93
- parser.add_argument("--session-id", help="Backfill one session only")
94
- args = parser.parse_args()
95
- result = run_backfill(args.session_id)
96
- print(f"Done: {result['actions_updated']}/{result['actions_total']} action rows updated")
97
-
98
-
99
- if __name__ == "__main__":
100
- main()
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
src/analysis.ts CHANGED
@@ -50,35 +50,6 @@ export function analysisRouter(db: SupabaseClient, secret: string, onCompleted?:
50
  next();
51
  });
52
  router.use(express.json({ limit: "2mb" }));
53
- router.post("/operations/playback-backfill", async (_req, res, next) => {
54
- try {
55
- if (res.locals.isAdmin !== true) return res.sendStatus(403);
56
- const response = await fetch(new URL("/internal/playback-backfill", env.ANALYSIS_API_URL), {
57
- method: "POST", headers: {
58
- "x-sporalize-proxy-secret": secret,
59
- "x-sporalize-user-id": res.locals.userId as string,
60
- "x-sporalize-admin": "true",
61
- }, signal: AbortSignal.timeout(15_000),
62
- });
63
- const body = await response.json().catch(() => ({}));
64
- return res.status(response.status).json(body);
65
- } catch (error) { next(error); }
66
- });
67
- router.get("/operations/playback-backfill/:jobId", async (req, res, next) => {
68
- try {
69
- if (res.locals.isAdmin !== true) return res.sendStatus(403);
70
- const jobId = z.uuid().parse(req.params.jobId);
71
- const response = await fetch(new URL(`/internal/playback-backfill/${jobId}`, env.ANALYSIS_API_URL), {
72
- headers: {
73
- "x-sporalize-proxy-secret": secret,
74
- "x-sporalize-user-id": res.locals.userId as string,
75
- "x-sporalize-admin": "true",
76
- }, signal: AbortSignal.timeout(15_000),
77
- });
78
- const body = await response.json().catch(() => ({}));
79
- return res.status(response.status).json(body);
80
- } catch (error) { next(error); }
81
- });
82
  router.post("/", async (req, res, next) => {
83
  try {
84
  const input = beginSchema.parse(req.body);
 
50
  next();
51
  });
52
  router.use(express.json({ limit: "2mb" }));
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
53
  router.post("/", async (req, res, next) => {
54
  try {
55
  const input = beginSchema.parse(req.body);