fix(viewer+backend): cache sorties TTL 10min + /sorties/local + viewer non-bloquant
This commit is contained in:
@@ -1,6 +1,7 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
import gzip
|
import gzip
|
||||||
import json
|
import json
|
||||||
|
import time
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import AsyncGenerator
|
from typing import AsyncGenerator
|
||||||
|
|
||||||
@@ -9,7 +10,7 @@ from fastapi.middleware.cors import CORSMiddleware
|
|||||||
from fastapi.responses import JSONResponse, StreamingResponse
|
from fastapi.responses import JSONResponse, StreamingResponse
|
||||||
|
|
||||||
from .config import OUTPUT_DIR
|
from .config import OUTPUT_DIR
|
||||||
from .runner import run_pipeline, scan_sorties
|
from .runner import run_pipeline, scan_sorties, scan_sorties_local
|
||||||
|
|
||||||
app = FastAPI(title="COSMA Pipeline Runner")
|
app = FastAPI(title="COSMA Pipeline Runner")
|
||||||
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
|
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
|
||||||
@@ -17,10 +18,51 @@ app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], all
|
|||||||
# Active pipeline jobs: sortie_id → asyncio.Queue
|
# Active pipeline jobs: sortie_id → asyncio.Queue
|
||||||
_jobs: dict[str, asyncio.Queue] = {}
|
_jobs: dict[str, asyncio.Queue] = {}
|
||||||
|
|
||||||
|
# Cache sorties avec TTL 10min
|
||||||
|
_sorties_cache: list | None = None
|
||||||
|
_sorties_cache_ts: float = 0.0
|
||||||
|
_SORTIES_TTL = 600.0 # 10 minutes
|
||||||
|
_sorties_refresh_lock: asyncio.Lock | None = None
|
||||||
|
|
||||||
|
|
||||||
|
def _get_lock() -> asyncio.Lock:
|
||||||
|
global _sorties_refresh_lock
|
||||||
|
if _sorties_refresh_lock is None:
|
||||||
|
_sorties_refresh_lock = asyncio.Lock()
|
||||||
|
return _sorties_refresh_lock
|
||||||
|
|
||||||
|
|
||||||
|
async def _refresh_sorties_cache() -> None:
|
||||||
|
"""Refresh cache in background (holds lock to avoid parallel rclone calls)."""
|
||||||
|
global _sorties_cache, _sorties_cache_ts
|
||||||
|
lock = _get_lock()
|
||||||
|
async with lock:
|
||||||
|
# Double-check after acquiring lock
|
||||||
|
if time.monotonic() - _sorties_cache_ts < _SORTIES_TTL:
|
||||||
|
return
|
||||||
|
result = await asyncio.to_thread(scan_sorties)
|
||||||
|
_sorties_cache = result
|
||||||
|
_sorties_cache_ts = time.monotonic()
|
||||||
|
|
||||||
|
|
||||||
@app.get("/sorties")
|
@app.get("/sorties")
|
||||||
async def list_sorties():
|
async def list_sorties():
|
||||||
return await scan_sorties()
|
global _sorties_cache, _sorties_cache_ts
|
||||||
|
now = time.monotonic()
|
||||||
|
if _sorties_cache is None:
|
||||||
|
# Premier appel: bloquant (cache vide)
|
||||||
|
await _refresh_sorties_cache()
|
||||||
|
elif now - _sorties_cache_ts >= _SORTIES_TTL:
|
||||||
|
# Cache périmé: retourne le cache, refresh en arrière-plan
|
||||||
|
asyncio.create_task(_refresh_sorties_cache())
|
||||||
|
return _sorties_cache or []
|
||||||
|
|
||||||
|
|
||||||
|
@app.get("/sorties/local")
|
||||||
|
async def list_sorties_local():
|
||||||
|
"""Scan /data/sorties local (NAS, instantané) sans rclone."""
|
||||||
|
sorties = await asyncio.to_thread(scan_sorties_local)
|
||||||
|
return sorties
|
||||||
|
|
||||||
|
|
||||||
@app.post("/run/{sortie_id:path}")
|
@app.post("/run/{sortie_id:path}")
|
||||||
|
|||||||
@@ -122,7 +122,7 @@ async def run_pipeline(sortie_id: str, queue: asyncio.Queue) -> None:
|
|||||||
|
|
||||||
|
|
||||||
async def scan_sorties() -> list[dict]:
|
async def scan_sorties() -> list[dict]:
|
||||||
"""List available sorties on GDrive via rclone lsd."""
|
"""List available sorties on GDrive via rclone lsd (lent ~30s)."""
|
||||||
result = subprocess.run(
|
result = subprocess.run(
|
||||||
["rclone", "lsd", GDRIVE_REMOTE],
|
["rclone", "lsd", GDRIVE_REMOTE],
|
||||||
capture_output=True, text=True
|
capture_output=True, text=True
|
||||||
@@ -135,3 +135,15 @@ async def scan_sorties() -> list[dict]:
|
|||||||
processed = (OUTPUT_DIR / name / "processed" / "usv.json.gz").exists()
|
processed = (OUTPUT_DIR / name / "processed" / "usv.json.gz").exists()
|
||||||
sorties.append({"id": name, "processed": processed})
|
sorties.append({"id": name, "processed": processed})
|
||||||
return sorties
|
return sorties
|
||||||
|
|
||||||
|
|
||||||
|
def scan_sorties_local() -> list[dict]:
|
||||||
|
"""List sorties already synced locally in OUTPUT_DIR (instantané, pas rclone)."""
|
||||||
|
if not OUTPUT_DIR.exists():
|
||||||
|
return []
|
||||||
|
sorties = []
|
||||||
|
for d in sorted(OUTPUT_DIR.iterdir()):
|
||||||
|
if d.is_dir():
|
||||||
|
processed = (d / "processed" / "usv.json.gz").exists()
|
||||||
|
sorties.append({"id": d.name, "processed": processed})
|
||||||
|
return sorties
|
||||||
|
|||||||
@@ -1121,29 +1121,60 @@ async function loadSortieData(sortieId) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async function loadSorties() {
|
function _wireSortieSelect() {
|
||||||
try {
|
const sel = document.getElementById('sortie-select');
|
||||||
const resp = await fetch(`${API2}/sorties`);
|
// Evite double-binding si appelé plusieurs fois
|
||||||
if (!resp.ok) return;
|
if (sel._wired) return;
|
||||||
const sorties = await resp.json();
|
sel._wired = true;
|
||||||
const sel = document.getElementById('sortie-select');
|
sel.addEventListener('change', () => {
|
||||||
sorties.forEach(s => {
|
const btn = document.getElementById('btn-sync');
|
||||||
const opt = document.createElement('option');
|
btn.disabled = !sel.value;
|
||||||
opt.value = s.id;
|
if (sel.value) {
|
||||||
opt.textContent = s.id + (s.processed ? ' ✓' : '');
|
const opt = sel.options[sel.selectedIndex];
|
||||||
sel.appendChild(opt);
|
if (opt.textContent.includes('✓')) {
|
||||||
});
|
loadSortieData(sel.value);
|
||||||
sel.addEventListener('change', () => {
|
|
||||||
const btn = document.getElementById('btn-sync');
|
|
||||||
btn.disabled = !sel.value;
|
|
||||||
if (sel.value) {
|
|
||||||
const opt = sel.options[sel.selectedIndex];
|
|
||||||
if (opt.textContent.includes('✓')) {
|
|
||||||
loadSortieData(sel.value);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
function _populateSortieSelect(sorties) {
|
||||||
|
const sel = document.getElementById('sortie-select');
|
||||||
|
// Vider sauf première option placeholder
|
||||||
|
while (sel.options.length > 1) sel.remove(1);
|
||||||
|
if (!sorties || !sorties.length) {
|
||||||
|
sel.options[0].textContent = '— Aucune sortie —';
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
sel.options[0].textContent = '— Sortie —';
|
||||||
|
sorties.forEach(s => {
|
||||||
|
const opt = document.createElement('option');
|
||||||
|
opt.value = s.id;
|
||||||
|
opt.textContent = s.id + (s.processed ? ' ✓' : '');
|
||||||
|
sel.appendChild(opt);
|
||||||
|
});
|
||||||
|
_wireSortieSelect();
|
||||||
|
}
|
||||||
|
|
||||||
|
function loadSorties() {
|
||||||
|
const sel = document.getElementById('sortie-select');
|
||||||
|
sel.options[0].textContent = 'Sorties: chargement…';
|
||||||
|
|
||||||
|
const controller = new AbortController();
|
||||||
|
const timeoutId = setTimeout(() => controller.abort(), 5000);
|
||||||
|
|
||||||
|
fetch(`${API2}/sorties`, { signal: controller.signal })
|
||||||
|
.then(resp => {
|
||||||
|
clearTimeout(timeoutId);
|
||||||
|
if (!resp.ok) throw new Error('HTTP ' + resp.status);
|
||||||
|
return resp.json();
|
||||||
|
})
|
||||||
|
.then(sorties => { _populateSortieSelect(sorties); })
|
||||||
|
.catch(e => {
|
||||||
|
clearTimeout(timeoutId);
|
||||||
|
sel.options[0].textContent = '— Pipeline indisponible —';
|
||||||
|
console.warn('loadSorties:', e.message);
|
||||||
});
|
});
|
||||||
} catch(e) { console.warn('pipeline-runner unavailable', e); }
|
|
||||||
}
|
}
|
||||||
|
|
||||||
document.getElementById('btn-sync').addEventListener('click', async () => {
|
document.getElementById('btn-sync').addEventListener('click', async () => {
|
||||||
|
|||||||
Reference in New Issue
Block a user