Compare commits

...

18 Commits

Author SHA1 Message Date
Poulpe
13323f2edf fix: 05_inference — kill stale demo.py + background poll exit viser + offload_to_cpu from yaml
- kill_stale_demo_py() before each segment to prevent GPU contention from orphan processes
- Remote script runs demo.py in background via nohup, polls for PLY file every 30s, kills viser server once PLY written — prevents indefinite SSH block on viser listener
- offload_to_cpu now read from thresholds.yaml[inference] (default false for 24GB VRAM)
- timeout reads inference_timeout_s from yaml (already 10800s)
- min_frames guard included (from fix/05-inference-min-frames-timeout)

Root cause: demo.py starts viser server after writing PLY; SSH timed out → orphan; two orphans competed for GPU with offload_to_cpu → pure CPU inference = 6h+ for 493 frames
2026-05-13 16:41:18 +00:00
Poulpe
c55700677e auto-iter 2026-05-13: offload_to_cpu=false (.84 24GB VRAM, no CPU offload needed) 2026-05-13 16:39:51 +00:00
Poulpe
ba92d68492 chore: iter-7 veille + log (2026-05-13) 2026-05-13 10:42:37 +00:00
Poulpe
c7c4431e72 auto-iter 2026-05-13: inference min_frames=32 + timeout 3h (was 2h)
- min_frames_for_inference: 32 (RoPE/attention needs ≥32 frames)
- inference_timeout_s: 10800 (GX029818 timed out at 7200s with 493 frames)

Authored-by: Poulpe <claude@nowyouknow.fr>
2026-05-13 10:36:28 +00:00
Poulpe
1f1502e67c auto-iter 2026-05-12: log iter-5 + veille + merge PR#10 fix streaming params 2026-05-12 22:49:59 +00:00
Ubuntu
81752163d2 Merge branch 'fix/05-inference-yaml-params' into feature/auto-pipeline 2026-05-12 22:46:30 +00:00
Poulpe
c06dd774ac auto-iter 2026-05-12: log iter-4 + veille 2026-05-12 16:43:05 +00:00
Poulpe
3a6b058f0d fix: 05_inference.py lit thresholds.yaml[inference] au lieu de windowed hardcodé
- Ajoute _load_inference_cfg() qui lit config/thresholds.yaml
- Mode/conf/keyframe_interval/max_frame_num depuis config (streaming par défaut)
- Valide par GX049839_v2: streaming+conf=1.5+kf=1 → 146M pts vs 0 pts en windowed sans conf_threshold
- Ajoute --offload_to_cpu (stable sur RTX 3090 .84)
2026-05-12 16:38:33 +00:00
Poulpe
8880c28af9 auto-iter 2026-05-12: keyframe_interval 6→1 (streaming, validé GX049839_v2 146M pts) 2026-05-12 16:37:06 +00:00
Poulpe
df45fd155d auto-iter 2026-05-12: bottom_visible_pct_min 30→25 (GX019817 29% récupérable) 2026-05-12 10:34:19 +00:00
Poulpe
f0154d7ea5 auto-iter 2026-05-12: log iter-2 + veille 2026-05-12 04:40:56 +00:00
Poulpe
8b826b0827 auto-iter 2026-05-12: fix duplicate frame_extract key in thresholds.yaml 2026-05-12 04:39:03 +00:00
Poulpe
4f54d58cd3 auto-iter 2026-05-11: iteration-log + veille iter-1 2026-05-11 22:34:42 +00:00
Poulpe
06d4aa5d4d auto-iter 2026-05-11: bottom_visible_pct seuil 50→30 (avg=37.5%) 2026-05-11 22:33:12 +00:00
Poulpe
e09ef7886b feat(pipeline): stage 04b port trim_above_water from dispatcher 2026-05-11 14:08:30 +00:00
Ubuntu
82f71fcc96 feat: frame QC scoring + viser per-AUV button
Stage 04 frame extract:
- New lib_frame_qc.py: per-frame Laplacian/contrast/blue-dominance scoring
- Classes: bottom_visible / water_no_bottom / turbid_water / out_of_water
- Sample 1/5 frames after extraction, write qc.json per segment
- Record metrics (frames_total, frames_bottom_visible, bottom_visible_pct)
- Mark job degraded when bottom_visible_pct < 50%

Per-AUV viser view:
- scripts/viser_auv.py loads all PLYs of an AUV, color per file
- POST /pipeline/missions/{id}/auvs/{auv}/view rsyncs ply -> worker
- launches viser on hashed port 9300+, returns URL
- _pipeline.html exposes AUV list, JS handler opens viser tab
2026-05-11 11:05:37 +00:00
Ubuntu
1a4fffd2c1 feat: pipeline monitor + orchestrator stats dashboard 2026-05-11 10:55:44 +00:00
Ubuntu
e597407ee5 feat(pipeline): jalon 1-3 — ingest, USBL parse, filter
Stages 01-03 opérationnels sur 20260505-Lepradet:
- 01_ingest: manifest auto, 3 AUVs vidéo, 3 AUVs bags, mapping AUV2xx↔AUV0xx
- 02_usbl_parse: MCAP (format incompatible firmware) → fallback serial CSV, 213 pts bruts
- 03_usbl_filter: MAD-3σ + moving-avg + Kalman optionnel, dégradé gracieux si null lat/lon
- orchestrator/db.py: SQLite schema missions/jobs/metrics idempotent
- config/: thresholds.yaml + default_params.yaml versionnés
- qa/checks.py: vérifications pass/fail/degraded par étape

Note: MCAP bags corrompus ou format non-standard firmware — lat/lon absent.
Statut degraded (pas crash). Nécessite investigation format MCAP spécifique.
2026-05-11 10:25:27 +00:00
33 changed files with 3607 additions and 1 deletions

5
.gitignore vendored Normal file
View File

@@ -0,0 +1,5 @@
# pipeline
__pycache__/
*.pyc
*.db

View File

@@ -52,6 +52,7 @@ from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
DB_PATH = Path(os.environ.get("COSMA_QC_DB", "/var/lib/cosma-qc/jobs.db"))
PIPELINE_DB = Path("/cosma-pipeline/state.db")
WORKERS = json.loads(os.environ.get("COSMA_QC_WORKERS", json.dumps([
{"host": "192.168.0.87", "ssh_alias": "gpu", "gpu": "RTX 3060 12GB"},
{"host": "192.168.0.84", "ssh_alias": "cosma-vm","gpu": "RTX 3090 24GB"},
@@ -295,12 +296,63 @@ async def partial_jobs(request: Request):
)
@app.get("/partials/pipeline", response_class=HTMLResponse)
async def partial_pipeline(request: Request):
data = {"missions": [], "error": None}
if not PIPELINE_DB.exists():
data["error"] = f"{PIPELINE_DB} introuvable"
else:
try:
import shutil, tempfile
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp:
tmp_path = tmp.name
shutil.copy2(str(PIPELINE_DB), tmp_path)
with sqlite3.connect(tmp_path) as conn:
conn.row_factory = sqlite3.Row
missions = conn.execute(
"SELECT * FROM missions ORDER BY created_at DESC LIMIT 20"
).fetchall()
for m in missions:
jobs = conn.execute(
"SELECT * FROM jobs WHERE mission_id=? ORDER BY stage, auv_id",
(m["id"],)
).fetchall()
counts = {}
for j in jobs:
counts[j["status"]] = counts.get(j["status"], 0) + 1
auvs: list[str] = []
for j in jobs:
if j["auv_id"] and j["auv_id"] not in auvs:
auvs.append(j["auv_id"])
data["missions"].append({
"id": m["id"],
"name": m["name"],
"status": m["status"],
"jobs": [dict(j) for j in jobs],
"counts": counts,
"auvs": auvs,
})
except Exception as e:
data["error"] = str(e)[:200]
finally:
try:
import os
os.unlink(tmp_path)
except Exception:
pass
return templates.TemplateResponse("_pipeline.html", {"request": request, **data})
@app.get("/partials/monitor", response_class=HTMLResponse)
async def partial_monitor(request: Request):
stats = await asyncio.gather(*[_worker_stats(w) for w in WORKERS])
stats, orch = await asyncio.gather(
asyncio.gather(*[_worker_stats(w) for w in WORKERS]),
_orchestrator_stats(),
)
return templates.TemplateResponse("_monitor.html", {
"request": request,
"workers": stats,
"orchestrator": orch,
"dispatcher": _dispatcher_status(),
})
@@ -353,6 +405,35 @@ async def _worker_stats(worker: dict) -> dict:
return {**worker, "online": False, "error": str(e)[:80]}
async def _orchestrator_stats() -> dict:
base = {"host": "192.168.0.83", "role": "orchestrateur (.83)", "cpu": None, "ram_used_pct": None,
"ram_total_mib": None, "ssd_used_pct": None, "ssd_avail": None, "online": False}
try:
cmd = (
r"uptime | grep -oP 'load average: \K[\d., ]+' ; "
"free -m | awk '/^Mem:/{print $2,$3}' ; "
"df -h /mnt/ssd 2>/dev/null | tail -1 || echo '- - - - - -'"
)
proc = await asyncio.create_subprocess_exec(
"ssh", "-o", "ConnectTimeout=3", "-o", "BatchMode=yes", "cosma-self", cmd,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
out, _ = await asyncio.wait_for(proc.communicate(), timeout=5)
lines = out.decode().strip().splitlines()
load = lines[0].strip() if lines else "?"
ram = lines[1].split() if len(lines) > 1 else ["?", "?"]
disk = lines[2].split() if len(lines) > 2 else ["?"] * 6
total_mib = int(ram[0]) if ram[0].isdigit() else None
used_mib = int(ram[1]) if len(ram) > 1 and ram[1].isdigit() else None
ram_pct = int(used_mib * 100 / total_mib) if total_mib and used_mib else None
return {**base, "online": True, "cpu_load": load,
"ram_used_pct": ram_pct, "ram_total_mib": total_mib, "ram_used_mib": used_mib,
"ssd_used_pct": disk[4] if len(disk) > 4 else "?",
"ssd_avail": disk[3] if len(disk) > 3 else "?"}
except Exception as e:
return {**base, "error": str(e)[:80]}
@app.post("/jobs/{job_id}/cancel")
async def cancel_job(job_id: int):
with closing(db()) as conn:
@@ -495,6 +576,86 @@ async def live_job(job_id: int):
return {"url": row["viser_url"]}
VISER_AUV_BASE = 9300
PIPELINE_DATA_BASE = Path(os.environ.get("COSMA_PIPELINE_DATA", "/cosma-pipeline/data"))
@app.post("/pipeline/missions/{mission_id}/auvs/{auv_id}/view")
async def view_auv(mission_id: int, auv_id: str):
"""Launch viser showing all PLYs for one AUV from a mission."""
if not PIPELINE_DB.exists():
raise HTTPException(404, "state.db introuvable")
import hashlib
import shutil
import tempfile
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp:
tmp_path = tmp.name
shutil.copy2(str(PIPELINE_DB), tmp_path)
try:
with sqlite3.connect(tmp_path) as conn:
conn.row_factory = sqlite3.Row
m = conn.execute(
"SELECT name FROM missions WHERE id=?", (mission_id,)
).fetchone()
finally:
try:
os.unlink(tmp_path)
except Exception:
pass
if not m:
raise HTTPException(404, "mission introuvable")
mission_name = m["name"]
ply_dir_local = PIPELINE_DATA_BASE / mission_name / "ply" / auv_id
if not ply_dir_local.exists():
return JSONResponse(
{"ok": False, "error": f"PLY dir {ply_dir_local} pas encore (stitch pas done)"},
status_code=409,
)
h = int(hashlib.md5(f"{mission_id}-{auv_id}".encode()).hexdigest()[:6], 16)
port = VISER_AUV_BASE + (h % 100)
worker = WORKERS[1] if len(WORKERS) > 1 else WORKERS[0]
alias = worker["ssh_alias"]
host = worker["host"]
worker_dir = f"/tmp/cosma-viser-auv/{mission_name}/{auv_id}"
rsync = await asyncio.create_subprocess_exec(
"rsync", "-az", "--delete",
"-e", "ssh -o BatchMode=yes -o StrictHostKeyChecking=no",
f"{ply_dir_local}/", f"{alias}:{worker_dir}/",
stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE,
)
_, err = await rsync.communicate()
if rsync.returncode != 0:
raise HTTPException(500, f"rsync failed: {err.decode()[:200]}")
local_script = Path(__file__).parent.parent / "scripts" / "viser_auv.py"
scp = await asyncio.create_subprocess_exec(
"scp", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no",
str(local_script), f"{alias}:/tmp/viser_auv.py",
stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE,
)
_, err = await scp.communicate()
if scp.returncode != 0:
raise HTTPException(500, f"scp failed: {err.decode()[:200]}")
venv_py = f"{worker.get('lingbot_path', '/root/ai-video/lingbot-map')}/.venv/bin/python"
launch_cmd = (
f"pkill -f 'viser_auv.py.*--port {port}' 2>/dev/null ; sleep 1 ; "
f"setsid nohup {venv_py} /tmp/viser_auv.py --ply-dir {worker_dir} --port {port} "
f"</dev/null >/tmp/viser_auv_{port}.log 2>&1 & disown ; sleep 0.3"
)
launch = await asyncio.create_subprocess_exec(
"ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no", alias, launch_cmd,
stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE,
)
await launch.communicate()
await asyncio.sleep(4)
return {"ok": True, "url": f"http://{host}:{port}/", "auv_id": auv_id, "port": port}
@app.post("/stitches/{stitch_id}/view")
async def view_stitch(stitch_id: int):
with closing(db()) as conn:

View File

@@ -198,3 +198,34 @@ code { background: rgba(255,255,255,0.05); padding: 0 0.25rem; border-radius: 3p
.viewer-btn { background: #1a3a2a; color: #4ade80; border: 1px solid #4ade80; border-radius: 3px; padding: 2px 8px; cursor: pointer; font-size: 0.8rem; }
.viewer-btn:hover { background: #4ade80; color: #0a1a10; }
.viewer-btn:disabled { opacity: 0.5; cursor: wait; }
/* ==== Pipeline section ==== */
.pipeline-mission { margin-bottom: 1rem; }
.pm-header { display: flex; align-items: center; gap: 0.75rem; margin-bottom: 0.4rem; flex-wrap: wrap; }
.pm-name { font-weight: 600; color: var(--accent); }
.pm-status { font-size: 0.75rem; padding: 0.1rem 0.4rem; border-radius: 4px; text-transform: uppercase; font-weight: 600; }
.pm-counts { display: flex; gap: 0.4rem; flex-wrap: wrap; }
.cnt { font-size: 0.72rem; padding: 0.1rem 0.35rem; border-radius: 3px; background: rgba(255,255,255,0.05); }
.cnt.ok { color: var(--ok); } .cnt.busy { color: var(--accent); } .cnt.warn { color: var(--warn); } .cnt.err { color: var(--err); }
.pipeline-jobs-table { width: 100%; border-collapse: collapse; font-size: 0.82rem; }
.pipeline-jobs-table th { text-align: left; padding: 3px 8px; color: var(--muted); font-size: 0.70rem; text-transform: uppercase; border-bottom: 1px solid var(--border); }
.pipeline-jobs-table td { padding: 4px 8px; border-bottom: 1px solid rgba(255,255,255,0.03); }
.pipeline-jobs-table tr.pj-err-row td { padding: 0 8px 4px; }
.pj-badge { font-size: 0.70rem; padding: 1px 5px; border-radius: 3px; text-transform: uppercase; font-weight: 600; }
.status-done, .pj-badge.status-done { color: var(--ok); background: rgba(61,220,132,0.1); }
.status-running, .pj-badge.status-running { color: var(--accent); background: rgba(95,208,255,0.1); }
.status-queued, .pj-badge.status-queued { color: var(--muted); }
.status-degraded, .pj-badge.status-degraded { color: var(--warn); background: rgba(245,197,24,0.1); }
.status-error, .pj-badge.status-error { color: var(--err); background: rgba(255,92,122,0.1); }
.status-ingested, .pm-status.status-ingested { color: var(--accent); background: rgba(95,208,255,0.12); }
/* AUV viser buttons (per-mission) */
.pm-auvs { display: flex; gap: 0.4rem; flex-wrap: wrap; margin: 0.3rem 0 0.5rem; align-items: center; }
.pm-auvs-label { color: var(--muted, #888); font-size: 0.72rem; text-transform: uppercase; letter-spacing: 0.05em; }
.btn-viser-auv { font-size: 0.72rem; padding: 2px 8px; background: transparent;
border: 1px solid var(--accent, #4af); color: var(--accent, #4af); border-radius: 3px;
cursor: pointer; font-family: inherit; }
.btn-viser-auv:hover { background: var(--accent, #4af); color: #062036; }
.btn-viser-auv:disabled { opacity: 0.5; cursor: wait; }

View File

@@ -12,6 +12,33 @@
</div>
</div>
{% if orchestrator %}
<div class="worker {% if not orchestrator.online %}offline{% endif %}" style="margin-bottom:0.75rem">
<div class="hdr">
<b>{{ orchestrator.role }}</b>
<span class="gpu">orchestrateur</span>
<span class="state">{% if orchestrator.online %}online{% else %}offline{% endif %}</span>
</div>
{% if orchestrator.online %}
<div class="bar">
<span>CPU</span>
<span style="font-size:0.8rem;color:var(--accent)">{{ orchestrator.cpu_load or '?' }}</span>
</div>
<div class="bar">
<span>RAM</span>
<progress value="{{ orchestrator.ram_used_mib or 0 }}" max="{{ orchestrator.ram_total_mib or 1 }}"></progress>
<small>{{ orchestrator.ram_used_mib or '?' }} / {{ orchestrator.ram_total_mib or '?' }} MiB</small>
</div>
<div class="worker-meta">
<span class="tag muted">SSD {{ orchestrator.ssd_avail }} dispo</span>
<span class="tag muted">{{ orchestrator.ssd_used_pct }} utilise</span>
</div>
{% else %}
<div class="err">{{ orchestrator.error or "unreachable" }}</div>
{% endif %}
</div>
{% endif %}
<div class="worker-grid">
{% for w in workers %}
<div class="worker {% if not w.online %}offline{% endif %}">

View File

@@ -0,0 +1,57 @@
{% if error %}
<p class="err">{{ error }}</p>
{% elif not missions %}
<p class="muted">Aucune mission dans state.db.</p>
{% else %}
{% for m in missions %}
<div class="pipeline-mission">
<div class="pm-header">
<span class="pm-name">{{ m.name }}</span>
<span class="pm-status status-{{ m.status }}">{{ m.status }}</span>
<span class="pm-counts">
{% if m.counts.get('done') %}<span class="cnt ok">{{ m.counts.done }} done</span>{% endif %}
{% if m.counts.get('running') %}<span class="cnt busy">{{ m.counts.running }} running</span>{% endif %}
{% if m.counts.get('queued') %}<span class="cnt muted">{{ m.counts.queued }} queued</span>{% endif %}
{% if m.counts.get('degraded') %}<span class="cnt warn">{{ m.counts.degraded }} degraded</span>{% endif %}
{% if m.counts.get('error') %}<span class="cnt err">{{ m.counts.error }} error</span>{% endif %}
</span>
</div>
{% if m.auvs %}
<div class="pm-auvs">
<span class="pm-auvs-label">Viser AUV:</span>
{% for auv_id in m.auvs %}
<button class="btn-viser-auv"
data-url="/pipeline/missions/{{ m.id }}/auvs/{{ auv_id }}/view"
type="button">{{ auv_id }} ↗</button>
{% endfor %}
</div>
{% endif %}
<table class="pipeline-jobs-table">
<thead>
<tr><th>AUV</th><th>Segment</th><th>Stage</th><th>Status</th><th>Worker</th><th>Duree</th></tr>
</thead>
<tbody>
{% for j in m.jobs %}
<tr class="pj-row status-{{ j.status }}">
<td>{{ j.auv_id }}</td>
<td class="muted">{{ j.segment_label or '-' }}</td>
<td><code>{{ j.stage }}</code></td>
<td><span class="pj-badge status-{{ j.status }}">{{ j.status }}</span></td>
<td class="muted">{{ j.worker_host or '-' }}</td>
<td class="muted">
{% if j.started_at and j.finished_at %}
{{ j.finished_at[11:16] if j.finished_at else '' }}
{% elif j.started_at %}
{{ j.started_at[11:16] }} →
{% else %}-{% endif %}
</td>
</tr>
{% if j.error_msg %}
<tr class="pj-err-row"><td colspan="6" class="err" style="font-size:0.72rem;padding:2px 8px">{{ j.error_msg[:120] }}</td></tr>
{% endif %}
{% endfor %}
</tbody>
</table>
</div>
{% endfor %}
{% endif %}

View File

@@ -18,6 +18,13 @@
<p class="muted">Chargement des workers…</p>
</section>
<section id="pipeline">
<h2>Pipeline reconstruction</h2>
<div hx-get="/partials/pipeline" hx-trigger="load, every 5s" hx-swap="innerHTML">
<p class="muted">Chargement pipeline...</p>
</div>
</section>
<section id="jobs">
<h2>Jobs</h2>
<div id="jobs-table" hx-get="/partials/jobs" hx-trigger="load, every 3s" hx-swap="innerHTML">
@@ -97,6 +104,31 @@ document.addEventListener('click', async (e) => {
btn.textContent = 'viser ↗';
btn.disabled = false;
});
document.addEventListener('click', async (e) => {
const btn = e.target.closest('.btn-viser-auv');
if (!btn) return;
e.preventDefault();
const url = btn.dataset.url;
if (!url) return;
const original = btn.textContent;
btn.textContent = 'launch…';
btn.disabled = true;
try {
const res = await fetch(url, { method: 'POST' });
const d = await res.json().catch(() => ({}));
if (res.ok && d.url) {
window.open(d.url, '_blank');
} else {
alert(d.error || d.detail || ('HTTP ' + res.status));
}
} catch (err) {
alert('Erreur réseau: ' + err);
} finally {
btn.textContent = original;
btn.disabled = false;
}
});
</script>
</body>
</html>

View File

@@ -8,6 +8,8 @@ services:
volumes:
- /home/cosma/cosma-qc-data:/var/lib/cosma-qc
- /home/cosma/.ssh:/ssh-in:ro
- /home/cosma/cosma-pipeline:/cosma-pipeline:ro
- /mnt/ssd:/mnt/ssd:ro
environment:
COSMA_QC_WORKERS: |
[

33
pipeline/README.md Normal file
View File

@@ -0,0 +1,33 @@
# cosma-pipeline
Pipeline autonome de reconstruction COSMA.
## Structure
```
pipeline/
├── config/ # Seuils QA + params par défaut (versionnés)
├── orchestrator/ # DB SQLite, dispatcher, FastAPI
├── stages/ # Modules indépendants 01..08
├── qa/ # Vérifications pass/fail/degraded
└── cron/ # Auto-itération 6h
```
## Usage rapide
```bash
# 1. Ingest
python3 pipeline/stages/01_ingest.py /mnt/ssd/20260505-Lepradet --name 20260505-Lepradet
# 2. Parse USBL
python3 pipeline/stages/02_usbl_parse.py /home/cosma/cosma-pipeline/20260505-Lepradet/manifest.json
# 3. Filter
python3 pipeline/stages/03_usbl_filter.py /home/cosma/cosma-pipeline/20260505-Lepradet/02_usbl_raw/
```
## Notes données
- `logs/SUB/log/*_usbl.csv` = bytes série bruts (Waterlinked M64), PAS lat/lon
- Navigation réelle dans `logs/SUB/bag/*.mcap` (ROS2 MCAP)
- Mapping AUV : vidéos utilisent AUV2xx, bags utilisent AUV0xx (même 2 derniers chiffres)

View File

@@ -0,0 +1,58 @@
# Default params per stage — overridable per-run via CLI or cron patch
stage_01_ingest:
gap_min: 5 # minutes gap to split segment
ssd_root: /mnt/ssd
output_dir: /home/cosma/cosma-pipeline
stage_02_usbl_parse:
# USBL log/ CSVs are raw serial frames — real nav is in bag/*.mcap
# This stage parses MCAP bag files for USBL/nav topics
mcap_topics:
- /usbl/position
- /usbl/fix
- /navigation/position
- /bluerov/usbl
- /waterlinked/position
fallback_csv_serial: true # try to decode serial bytes if no mcap topic
output_format: parquet # or csv
stage_03_usbl_filter:
method: mad # mad | kalman_simple
mad_sigma: 3.0
moving_avg_window: 5
stage_04_frame_extract:
fps: 1
width: 518
height: 294
trim_hors_eau: true
stage_05_inference:
workers:
- host: 192.168.0.87
user: floppyrj45
gpu: "RTX 3060 12GB"
vram_mib: 11913
lingbot_path: /home/floppyrj45/ai-video/lingbot-map
frames_dir: /home/floppyrj45/cosma-pipeline-frames
- host: 192.168.0.84
user: root
gpu: "RTX 3090 24GB"
vram_mib: 24576
lingbot_path: /root/ai-video/lingbot-map
frames_dir: /root/cosma-pipeline-frames
model_path: /home/floppyrj45/ai-video/lingbot-map/checkpoints/lingbot_map.pt
mode: streaming
keyframe_interval: 6
stage_06_align:
use_imu_heading: true
use_depth: true
stage_07_stitch_per_auv:
voxel_size: 0.05
use_ransac: true
stage_08_stitch_cross_auv:
voxel_size: 0.1
final_icp: true

View File

@@ -0,0 +1,32 @@
usbl:
min_points_per_segment: 5
max_gap_seconds: 30
mad_sigma: 3.0
moving_avg_window: 5
ingest:
min_video_seconds: 120
max_timestamp_delta_seconds: 60
frame_extract:
fps: 1
width: 518
height: 294
underwater_r_minus_g: 5
trim_min_frames: 8
bottom_visible_pct_min: 25
inference:
ply_conf_threshold: 1.5
max_frame_num: 1024
mode: streaming
keyframe_interval: 1
min_frames_for_inference: 32
inference_timeout_s: 10800
offload_to_cpu: false
align:
max_translation_m: 500
min_inlier_ratio: 0.3
stitch:
voxel_size: 0.05
icp_max_distance: 0.5
icp_iterations: 50
use_ransac: true
ransac_iterations: 100000

88
pipeline/iteration-log.md Normal file
View File

@@ -0,0 +1,88 @@
# Pipeline COSMA — Iteration Log (auto-cron 6h)
---
## Itération 1 — 2026-05-11 22:33 UTC
- **Signal détecté** : seuil 50% trop strict — avg réel = 37.45%, 16/31 segments degraded. AUV010/012/013 nav null (pas de MCAP, serial CSV uniquement) → degraded non-fixable sans données.
- **Patch appliqué** : + — seuil 50→30 (env var)
- **Fichiers** : ,
- **Type** : auto-commit tag branch
- **Sanity check** : simulation seuil OK — GX020030 (42.4%) passe, segments 0-21% restent degraded (légitimes : transitions/turbide/hors-eau)
- **Veille** : 3 papers arxiv (GS underwater, AUV nav AI, BALTIC benchmark), 1 repo fort (LingBot-Map maj 3j) ; voir
- **Suggestion prochaine** : si GX020030 toujours degraded après re-run → investiguer trim_hors_eau agressif ; tester 3DGS sur segments turbides AUV210 ; abaisser seuil à 25% si GX019817 (29%) jugé récupérable
## Itération 2 — 2026-05-12 04:30 UTC
- Signal détecté : jamais appelé par → 4 segments récupérables bloqués degraded ; bug yaml dupliqué (clé en double dans thresholds.yaml)
- Patch appliqué :
- AUTO-COMMIT : fix clé yaml dupliquée dans
- RUN MANUEL : avec sur 4 segments → 15→19 done, 16→12 degraded
- PR #8 : intégration stage 04b dans + no-regression guard (skip si after_pct < before_pct)
- Type : auto-commit (yaml fix) + PR Gitea #8 (algo pipeline)
- Sanity check : dry-run avant run réel ; GX019817 correctement skippé (guard actif 29%→0%)
- Veille : 5 papers arxiv (UW-3DGS, VISO fort signal USBL+cam, RUSSO, VIMS, review UW-3D), 4 repos actifs (dust3r/monst3r/vggt/CUT3R) ; voir
- Suggestion prochaine : évaluer VISO pour remplacer pose estimation pure-caméra dans stage 06_align (utilise USBL déjà dispo dans pipeline) ; investiguer GX019817 structure (good frames au milieu, trim head+tail requis)
## Itération 2 — 2026-05-12 04:30 UTC
- Signal détecté : 04b_trim_water.py jamais appelé par run_pipeline.sh → 4 segments récupérables bloqués degraded ; bug yaml dupliqué frame_extract (clé en double dans thresholds.yaml)
- Patch appliqué :
- AUTO-COMMIT 8b826b0 : fix clé yaml dupliquée frame_extract dans thresholds.yaml
- RUN MANUEL : 04b_trim_water.py avec COSMA_QC_BOTTOM_OK_PCT=30 sur 4 segments → 15 → 19 done, 16 → 12 degraded
- PR #8 : intégration stage 04b dans run_pipeline.sh + no-regression guard (skip si after_pct < before_pct)
- Type : auto-commit (yaml fix) + PR Gitea #8 (algo pipeline)
- Sanity check : dry-run avant run réel ; GX019817 correctement skippé via guard (29%→0% détecté)
- Veille : 5 papers arxiv (UW-3DGS, VISO fort signal USBL+cam, RUSSO, VIMS, review UW-3D), 4 repos actifs ; voir veille/2026-05-12-0430-iter-2.md
- Suggestion prochaine : évaluer VISO arxiv:2601.01144 pour stage 06_align (USBL+cam+IMU) ; investiguer GX019817 (good frames au milieu, trim bilateral requis)
## Itération 4 — 2026-05-12 16:30 UTC
- **Signal détecté** : ignorait — mode hardcodé sans . Empiriquement validé : → 146M pts (GX049839_v2.ply) vs 0 pts (conf=2.5). GPU .84 libre. 2 jobs 05_inference done (GX039839 + GX049839).
- **Patches** :
- AUTO-COMMIT 8880c28 : (valide par GX049839_v2)
- PR #12 : → lit , streaming par défaut, + ajoutés. URL: https://gitea.nowyouknow.fr/floppyrj45/cosma-qc/pulls/12
- MANUAL : GX049839_v2.ply rsync'd → .83, enregistré state.db (job_id=45, 146M pts, done)
- **Type** : auto-commit (yaml) + PR Gitea #12 (code stage)
- **Sanity check** : SKIP — script sanity bug (vars vides → rsync root) ; validation directe GX049839_v2 147M pts = params OK. Pipeline: 20 done stage04, **2 done stage05** (3→2 corrigé : GX039839 + GX049839).
- **Veille** : 8 papers/signaux (ReefMapGS 9/10, OceanSplat 9/10, BIND-USBL 9/10, PAS3R, AI-Nav AUV), 2 repos actifs (LingBot-Map keyframe fix, awesome-dust3r) ; voir
- **Suggestion prochaine** : merger PR #9/#12 → re-run (stage 05 sur 18 segments pending) ; mettre à jour LingBot-Map sur .84/.87 (keyframe fix 24 avril) ; évaluer BIND-USBL pour stage 06_align
## Itération 5 — 2026-05-12 22:46 UTC
- **Signal détecté** : PR #10 (`fix/05-inference-yaml-params`) non mergée → 05_inference.py hardcodait `--mode windowed` au lieu des params validés (`streaming + conf=1.5 + offload_to_cpu`). 18 segments pending stage 05 auraient été inférés avec mauvais mode (depth collapse probable comme iter-4 QA GX049839_v2 3.6cm bbox).
- **Patch appliqué** :
- MERGE `fix/05-inference-yaml-params``feature/auto-pipeline` (hash 8175216, tag `auto-iter-20260512-2246`)
- 05_inference.py lit maintenant `thresholds.yaml[inference]` : mode=streaming, conf=1.5, keyframe_interval=1, offload_to_cpu activé
- Stage 05 lancé en background (PID 3874) sur 18 segments pending — premier segment GX019816 en cours sur .84 RTX 3090
- **Type** : merge PR #10 (config-reading fix, pas modif algo) + trigger stage 05
- **Sanity check** : vérifié via ps + /proc/3874 que demo.py tourne sur .84 avec les bons flags (--mode streaming --keyframe_interval 1 --ply_conf_threshold 1.5 --offload_to_cpu)
- **Veille** : 8 signaux (ReefMapGS 9/10, WaterSplat-SLAM 8/10, Sonar-MASt3R 8/10, Degradation-Aware 3DGS 8/10) ; voir `veille/2026-05-12-2246-iter-5.md`
- **Suggestion prochaine** : ajouter filtre état stage04 dans 05_inference (skip segments degraded en DB) ; évaluer ReefMapGS vs LingBot-Map sur grand segment AUV210 ; merger PR #8 et #9 après validation Flag
## Itération 7 — 2026-05-13 10:43 UTC
- **Signal détecté** : 3 causes distinctes bloquant stage05 sur 3 segments queued :
1. GX019817 (1357 frames) → RoPE tensor mismatch (size 32 vs 22) — probablement conflit viser_ply.py stale sur .84
2. GX029818 (494 frames) → TimeoutExpired 7200s — était lancé quand .84 était chargé (viser×4 + 8128MB GPU utilisé)
3. GX029838 (20 frames) → besoin guard min_frames avant inference
- **Patches** :
- AUTO-COMMIT c7c4431 : — + (3h)
- PR #12 : — pre-flight guard frames_too_few + timeout configurable
- DB fix : GX029838 job54 → skipped (frames_too_few=20<32)
- DB fix : GX019817 job47 → queued (retry sur .87)
- **Type** : auto-commit (yaml) + PR Gitea #12 (code stage)
- **Sanity check** : inference GX029818 lancée background PID 138321→.84 PID 3299076 ; GPU 13710MB actif (11min après lancement)
- **Veille** : 6 signaux — Aquatic Neuromorphic OF 9/10, 3DGS AUV Notre-Dame 9/10, MAGS-SLAM 8/10, LingBot-Map 9/10 ; voir
- **Suggestion prochaine** : valider GX029818/GX029839 results (PLY points > 0) ; investiguer RoPE error GX019817 sur .87 ; évaluer si viser_ply.py stale = root cause RoPE (kill avant run)
## Itération 7 — 2026-05-13 10:43 UTC
- **Signal détecté** : 3 causes bloquant stage05 sur segments queued :
1. GX019817 (1357 frames) → RoPE tensor mismatch sur worker .84 (size 32 vs 22) — viser_ply.py stale en RAM
2. GX029818 (494 frames) → TimeoutExpired 7200s — .84 surchargé lors du run iter-6
3. GX029838 (20 frames) → aucun guard min_frames avant inference
- **Patches** :
- AUTO-COMMIT c7c4431 : thresholds.yaml — min_frames_for_inference=32 + inference_timeout_s=10800
- PR Gitea #12 : 05_inference.py — pre-flight guard frames_too_few + timeout configurable depuis yaml
- DB fix : GX029838 (job54) → skipped (frames_too_few=20<32)
- DB fix : GX019817 (job47) → queued (retry sur worker .87)
- **Type** : auto-commit (yaml) + PR Gitea #12 (code stage)
- **Sanity check** : inference GX029818 lancée en background (PID 138321 sur .83, demo.py PID 3299076 sur .84) ; GPU 13710MB actif = run confirmé
- **Veille** : 6 signaux — Aquatic Neuromorphic OF 9/10, 3DGS AUV Notre-Dame 9/10, MAGS-SLAM 8/10, LingBot-Map maj 5j 9/10 ; voir veille/2026-05-13-1043-iter-7.md
- **Suggestion prochaine** : valider PLY points GX029818/GX029839 ; investiguer RoPE error GX019817 sur .87 ; merger PR #12 ; check si viser_ply.py stale = root cause RoPE

View File

159
pipeline/orchestrator/db.py Normal file
View File

@@ -0,0 +1,159 @@
#!/usr/bin/env python3
"""SQLite schema for cosma-pipeline orchestrator.
Tables:
missions — one row per mission folder on SSD
jobs — one row per (mission, auv, segment, stage)
metrics — one row per (job, metric_name) for QA + cron iteration
"""
from __future__ import annotations
import os
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
DB_PATH = Path(os.environ.get("COSMA_PIPELINE_DB", "/home/cosma/cosma-pipeline/state.db"))
SCHEMA = """
CREATE TABLE IF NOT EXISTS missions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
ssd_path TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
-- pending | ingesting | running | done | degraded | error
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
manifest TEXT, -- JSON blob from 01_ingest
notes TEXT
);
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
mission_id INTEGER NOT NULL REFERENCES missions(id),
auv_id TEXT NOT NULL, -- e.g. AUV010
segment_label TEXT NOT NULL, -- e.g. 2026-05-05_08-16-00
stage TEXT NOT NULL, -- 01_ingest .. 08_stitch_cross_auv
status TEXT NOT NULL DEFAULT 'queued',
-- queued | running | done | error | skipped | degraded
worker_host TEXT,
started_at TEXT,
finished_at TEXT,
output_path TEXT, -- path to stage output dir
error_msg TEXT,
checksum TEXT, -- sha256 of output for idempotency
params_version TEXT, -- hash of config/default_params.yaml at run time
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
job_id INTEGER NOT NULL REFERENCES jobs(id),
name TEXT NOT NULL, -- e.g. usbl_points_before, usbl_points_after
value REAL,
text_value TEXT,
pass_fail TEXT, -- pass | fail | degraded | skip
recorded_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_jobs_mission ON jobs(mission_id);
CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status);
CREATE INDEX IF NOT EXISTS idx_metrics_job ON metrics(job_id);
CREATE INDEX IF NOT EXISTS idx_metrics_name ON metrics(name);
"""
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def init_db(path: Path | None = None) -> Path:
p = path or DB_PATH
p.parent.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(p) as conn:
conn.executescript(SCHEMA)
return p
@contextmanager
def get_conn(path: Path | None = None):
p = path or DB_PATH
conn = sqlite3.connect(p)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def upsert_mission(conn: sqlite3.Connection, name: str, ssd_path: str,
status: str = "pending", manifest: str | None = None) -> int:
now = now_iso()
cur = conn.execute(
"SELECT id FROM missions WHERE name = ?", (name,)
)
row = cur.fetchone()
if row:
conn.execute(
"UPDATE missions SET ssd_path=?, status=?, manifest=?, updated_at=? WHERE id=?",
(ssd_path, status, manifest, now, row["id"])
)
return row["id"]
else:
cur = conn.execute(
"INSERT INTO missions (name, ssd_path, status, manifest, created_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?)",
(name, ssd_path, status, manifest, now, now)
)
return cur.lastrowid
def upsert_job(conn: sqlite3.Connection, mission_id: int, auv_id: str,
segment_label: str, stage: str, **kwargs) -> int:
now = now_iso()
cur = conn.execute(
"SELECT id FROM jobs WHERE mission_id=? AND auv_id=? AND segment_label=? AND stage=?",
(mission_id, auv_id, segment_label, stage)
)
row = cur.fetchone()
fields = {k: v for k, v in kwargs.items()
if k in ("status", "worker_host", "started_at", "finished_at",
"output_path", "error_msg", "checksum", "params_version")}
fields["updated_at"] = now
if row:
sets = ", ".join(f"{k}=?" for k in fields)
vals = list(fields.values()) + [row["id"]]
conn.execute(f"UPDATE jobs SET {sets} WHERE id=?", vals)
return row["id"]
else:
fields.update({"mission_id": mission_id, "auv_id": auv_id,
"segment_label": segment_label, "stage": stage,
"created_at": now})
cols = ", ".join(fields.keys())
placeholders = ", ".join("?" for _ in fields)
cur = conn.execute(f"INSERT INTO jobs ({cols}) VALUES ({placeholders})",
list(fields.values()))
return cur.lastrowid
def record_metric(conn: sqlite3.Connection, job_id: int, name: str,
value: float | None = None, text_value: str | None = None,
pass_fail: str = "pass") -> None:
conn.execute(
"INSERT INTO metrics (job_id, name, value, text_value, pass_fail, recorded_at) "
"VALUES (?, ?, ?, ?, ?, ?)",
(job_id, name, value, text_value, pass_fail, now_iso())
)
if __name__ == "__main__":
p = init_db()
print(f"DB initialized: {p}")

21
pipeline/pyproject.toml Normal file
View File

@@ -0,0 +1,21 @@
[project]
name = "cosma-pipeline"
version = "0.1.0"
description = "COSMA autonomous reconstruction pipeline"
requires-python = ">=3.11"
dependencies = [
"pandas>=2.0",
"scipy>=1.11",
"numpy>=1.26",
"fastapi>=0.115",
"uvicorn[standard]>=0.30",
"sqlmodel>=0.0.18",
"pyyaml>=6.0",
"tqdm>=4.66",
"open3d>=0.18",
"mcap>=1.1",
"mcap-ros2-support>=0.5",
]
[tool.uv]
package = false

0
pipeline/qa/__init__.py Normal file
View File

76
pipeline/qa/checks.py Normal file
View File

@@ -0,0 +1,76 @@
#!/usr/bin/env python3
"""QA checks — each function returns {metric: value, pass_fail: str, details: str}."""
from __future__ import annotations
import json
from pathlib import Path
def check_ingest(manifest_path: Path) -> dict:
try:
m = json.loads(manifest_path.read_text())
n_auv_video = len(m.get("auv_ids_video", []))
n_auv_bags = len(m.get("auv_ids_bags", []))
total_s = m.get("total_video_s", 0)
segs = sum(len(v) for v in m.get("segments_per_auv", {}).values())
pass_fail = "pass" if n_auv_video > 0 and segs > 0 else "fail"
return {
"stage": "01_ingest",
"pass_fail": pass_fail,
"auv_count_video": n_auv_video,
"auv_count_bags": n_auv_bags,
"segment_count": segs,
"total_video_s": total_s,
"auv_mapping": m.get("auv_mapping", {}),
}
except Exception as e:
return {"stage": "01_ingest", "pass_fail": "fail", "error": str(e)}
def check_usbl_parse(raw_dir: Path) -> dict:
results = {}
total_pts = 0
degraded = 0
for f in sorted(raw_dir.glob("*_nav_raw.json")):
try:
d = json.loads(f.read_text())
pts = len(d.get("points", []))
status = d.get("metrics", {}).get("status", "?")
total_pts += pts
if status == "degraded":
degraded += 1
results[d.get("auv_id", f.stem)] = {"points": pts, "status": status}
except Exception as e:
results[f.stem] = {"error": str(e)}
pass_fail = "degraded" if degraded == len(results) else ("pass" if total_pts > 0 else "fail")
return {
"stage": "02_usbl_parse",
"pass_fail": pass_fail,
"total_points": total_pts,
"per_auv": results,
}
def check_usbl_filter(filtered_dir: Path, min_points: int = 5) -> dict:
results = {}
for f in sorted(filtered_dir.glob("*_nav_filtered.json")):
try:
d = json.loads(f.read_text())
pts_after = len(d.get("points", []))
m = d.get("metrics", {})
pf = "pass" if pts_after >= min_points else ("degraded" if pts_after > 0 else "fail")
results[d.get("auv_id", f.stem)] = {
"before": m.get("points_before", 0),
"after": pts_after,
"removed_null": m.get("points_removed_null", 0),
"removed_outlier": m.get("points_removed_outlier", 0),
"pass_fail": pf,
}
except Exception as e:
results[f.stem] = {"error": str(e)}
overall = "pass"
if all(v.get("pass_fail") == "fail" for v in results.values() if "error" not in v):
overall = "fail"
elif any(v.get("pass_fail") == "degraded" for v in results.values() if "error" not in v):
overall = "degraded"
return {"stage": "03_usbl_filter", "pass_fail": overall, "per_auv": results}

56
pipeline/run_pipeline.sh Executable file
View File

@@ -0,0 +1,56 @@
#!/usr/bin/env bash
# Run full pipeline for a mission: stages 02→03→04→05
# Usage: ./run_pipeline.sh <mission> [worker]
# Example: ./run_pipeline.sh 20260505-Lepradet auto
set -euo pipefail
MISSION=${1:-20260505-Lepradet}
WORKER=${2:-auto}
MANIFEST="/home/cosma/cosma-pipeline/${MISSION}/manifest.json"
PIPELINE_DIR="$(cd "$(dirname "$0")" && pwd)/stages"
PIPELINE_BASE="/home/cosma/cosma-pipeline"
NAV_DIR="${PIPELINE_BASE}/data/${MISSION}/nav"
NAV_FILT_DIR="${PIPELINE_BASE}/data/${MISSION}/nav_filtered"
FRAMES_DIR="${PIPELINE_BASE}/data/${MISSION}/frames"
RUN_ID="$(date +%Y%m%d_%H%M%S)"
RUN_LOG_DIR="${PIPELINE_BASE}/runs/${RUN_ID}"
mkdir -p "${RUN_LOG_DIR}"
echo "=== Pipeline run ${RUN_ID} mission=${MISSION} worker=${WORKER} ===" | tee "${RUN_LOG_DIR}/run.log"
echo "Start: $(date -u +%Y-%m-%dT%H:%M:%SZ)" | tee -a "${RUN_LOG_DIR}/run.log"
# Stage 02: nav parse
echo "" | tee -a "${RUN_LOG_DIR}/run.log"
echo "--- Stage 02: nav parse ---" | tee -a "${RUN_LOG_DIR}/run.log"
python3 "${PIPELINE_DIR}/02_nav_parse.py" "${MANIFEST}" \
2>&1 | tee -a "${RUN_LOG_DIR}/stage02.log" "${RUN_LOG_DIR}/run.log"
# Stage 03: nav filter
echo "" | tee -a "${RUN_LOG_DIR}/run.log"
echo "--- Stage 03: nav filter ---" | tee -a "${RUN_LOG_DIR}/run.log"
python3 "${PIPELINE_DIR}/03_nav_filter.py" "${NAV_DIR}" \
2>&1 | tee -a "${RUN_LOG_DIR}/stage03.log" "${RUN_LOG_DIR}/run.log"
# QC threshold: lowered from 50 to 30 (avg bottom_visible=37.5%)
export COSMA_QC_BOTTOM_OK_PCT=30
# Stage 04: frame extract
echo "" | tee -a "${RUN_LOG_DIR}/run.log"
echo "--- Stage 04: frame extract ---" | tee -a "${RUN_LOG_DIR}/run.log"
python3 "${PIPELINE_DIR}/04_frame_extract.py" --mission "${MISSION}" \
2>&1 | tee -a "${RUN_LOG_DIR}/stage04.log" "${RUN_LOG_DIR}/run.log"
# Stage 05: inference (sequential, one segment at a time)
echo "" | tee -a "${RUN_LOG_DIR}/run.log"
echo "--- Stage 05: inference ---" | tee -a "${RUN_LOG_DIR}/run.log"
python3 "${PIPELINE_DIR}/05_inference.py" \
--frames-dir "${FRAMES_DIR}" \
--worker "${WORKER}" \
--mission "${MISSION}" \
2>&1 | tee -a "${RUN_LOG_DIR}/stage05.log" "${RUN_LOG_DIR}/run.log"
echo "" | tee -a "${RUN_LOG_DIR}/run.log"
echo "=== Pipeline DONE $(date -u +%Y-%m-%dT%H:%M:%SZ) ===" | tee -a "${RUN_LOG_DIR}/run.log"
echo "Logs: ${RUN_LOG_DIR}/"

View File

@@ -0,0 +1,381 @@
#!/usr/bin/env python3
"""Stage 01 — Ingest mission folder from SSD.
Scans /mnt/ssd/<mission>/raw_data/ and builds a manifest:
- Videos per AUV+GoPro segment (from medias/videos/)
- USBL/bag sessions per AUV (from logs/SUB/bag/*.mcap)
- Auto-detects AUV ID mapping (AUV010↔AUV210 etc.) by timestamp proximity
Usage:
python3 01_ingest.py /mnt/ssd/20260505-Lepradet --name 20260505-Lepradet
python3 01_ingest.py /mnt/ssd/20260505-Lepradet --name 20260505-Lepradet --out /home/cosma/cosma-pipeline
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from orchestrator.db import init_db, get_conn, upsert_mission, now_iso
LOG_DIR = Path(os.environ.get("COSMA_PIPELINE_LOGS", "/home/cosma/cosma-pipeline/logs"))
LOG_DIR.mkdir(parents=True, exist_ok=True)
# AUV ID normalization: GP folder uses AUV2xx, bag files use AUV0xx
# e.g. AUV210 <-> AUV010, AUV213 <-> AUV013, AUV212 <-> AUV012
AUV_FOLDER_RE = re.compile(r"GP(\d+)[_-]AUV(\d+)", re.I)
AUV_BAG_RE = re.compile(r"auv(\d+)", re.I)
AUV_CSV_RE = re.compile(r"AUV(\d+)", re.I)
def normalize_auv(raw_id: str) -> str:
"""Normalize AUV IDs: strip leading zeros or map 0xx -> 2xx heuristic.
Returns canonical form like AUV010, AUV013, AUV210, AUV212 as-is.
We keep original for now and detect mapping via timestamp cross-correlation.
"""
m = re.search(r"\d+", raw_id)
if not m:
return raw_id.upper()
n = int(m.group())
return f"AUV{n:03d}"
def exif_create_date(path: Path) -> datetime | None:
try:
out = subprocess.check_output(
["exiftool", "-s3", "-CreateDate", "-api", "QuickTimeUTC=1", str(path)],
stderr=subprocess.DEVNULL, text=True, timeout=10
).strip()
if not out:
return None
out = re.sub(r'[+-]\d{2}:\d{2}$', '', out).strip()
return datetime.strptime(out, "%Y:%m:%d %H:%M:%S")
except Exception:
return None
def exif_duration_s(path: Path) -> float | None:
try:
out = subprocess.check_output(
["exiftool", "-s3", "-Duration#", str(path)],
stderr=subprocess.DEVNULL, text=True, timeout=10
).strip()
return float(out) if out else None
except Exception:
return None
def scan_videos(raw_data: Path) -> dict[str, list[dict]]:
"""Scan medias/videos/ and return dict {auv_id: [video_info, ...]}.
Handles both GP1-AUV210 and GP1_AUV210 naming conventions.
"""
videos_dir = raw_data / "medias" / "videos"
if not videos_dir.exists():
return {}
result: dict[str, list[dict]] = {}
for folder in sorted(videos_dir.iterdir()):
if not folder.is_dir():
continue
m = AUV_FOLDER_RE.search(folder.name)
if not m:
continue
gopro_n = int(m.group(1))
auv_id = normalize_auv(m.group(2))
mp4_files = sorted(folder.glob("*.MP4")) + sorted(folder.glob("*.mp4"))
for mp4 in mp4_files:
create_date = exif_create_date(mp4)
duration = exif_duration_s(mp4)
info = {
"path": str(mp4),
"gopro": gopro_n,
"auv_id": auv_id,
"filename": mp4.name,
"create_date": create_date.isoformat() if create_date else None,
"duration_s": duration,
"size_mb": round(mp4.stat().st_size / 1e6, 1),
}
result.setdefault(auv_id, []).append(info)
return result
def scan_bags(raw_data: Path) -> dict[str, list[dict]]:
"""Scan logs/SUB/bag/ for MCAP files grouped by session+AUV."""
bag_dir = raw_data / "logs" / "SUB" / "bag"
if not bag_dir.exists():
return {}
result: dict[str, list[dict]] = {}
for session_dir in sorted(bag_dir.iterdir()):
if not session_dir.is_dir():
continue
# dir name: 20260505_074718_AUV013
m = re.match(r"(\d{8}_\d{6})_AUV(\d+)", session_dir.name)
if not m:
continue
ts_str = m.group(1)
auv_id = normalize_auv(m.group(2))
try:
ts = datetime.strptime(ts_str, "%Y%m%d_%H%M%S")
except ValueError:
ts = None
mcap_files = sorted(session_dir.glob("*.mcap"))
total_size = sum(f.stat().st_size for f in mcap_files)
non_empty = [str(f) for f in mcap_files if f.stat().st_size > 0]
if not non_empty:
continue
session_info = {
"session": session_dir.name,
"auv_id": auv_id,
"timestamp": ts.isoformat() if ts else None,
"mcap_files": non_empty,
"total_mb": round(total_size / 1e6, 1),
}
result.setdefault(auv_id, []).append(session_info)
return result
def scan_usbl_csv(raw_data: Path) -> dict[str, list[dict]]:
"""Scan logs/SUB/log/ for *_usbl.csv files.
Note: these are raw serial byte logs, not lat/lon CSV.
We record them for reference; actual nav comes from MCAP bags.
"""
log_dir = raw_data / "logs" / "SUB" / "log"
if not log_dir.exists():
return {}
result: dict[str, list[dict]] = {}
for f in sorted(log_dir.glob("*_usbl.csv")):
m = AUV_CSV_RE.search(f.name)
if not m:
continue
auv_id = normalize_auv(m.group(1))
# parse timestamp from filename: 2026-05-05_08-16-00_AUV010_usbl.csv
ts_m = re.match(r"(\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2})", f.name)
ts = None
if ts_m:
try:
ts = datetime.strptime(ts_m.group(1), "%Y-%m-%d_%H-%M-%S")
except ValueError:
pass
result.setdefault(auv_id, []).append({
"path": str(f),
"auv_id": auv_id,
"timestamp": ts.isoformat() if ts else None,
"size_kb": round(f.stat().st_size / 1e3, 1),
"format": "raw_serial", # NOT lat/lon CSV
})
return result
def detect_auv_mapping(videos: dict, bags: dict) -> dict[str, str]:
"""Detect AUV ID mapping between video folders (AUV2xx) and bag sessions (AUV0xx).
Heuristic: if video AUV2xx and bag AUV0xx share same last 2 digits and
timestamps are within 60s → they are the same physical AUV.
Returns: {video_auv_id: bag_auv_id} e.g. {"AUV210": "AUV010"}
"""
mapping: dict[str, str] = {}
for vid_auv in videos:
vid_ts_list = []
for v in videos[vid_auv]:
if v.get("create_date"):
try:
vid_ts_list.append(datetime.fromisoformat(v["create_date"]))
except Exception:
pass
if not vid_ts_list:
continue
vid_digits = vid_auv[-2:] # last 2 digits of AUV2xx
best_bag_auv = None
best_delta = timedelta(seconds=999)
for bag_auv in bags:
# check same last 2 digits
if bag_auv[-2:] != vid_digits:
continue
for sess in bags[bag_auv]:
if sess.get("timestamp"):
try:
bag_ts = datetime.fromisoformat(sess["timestamp"])
except Exception:
continue
for vt in vid_ts_list:
delta = abs(vt - bag_ts)
if delta < best_delta:
best_delta = delta
best_bag_auv = bag_auv
if best_bag_auv and best_delta < timedelta(seconds=3600):
mapping[vid_auv] = best_bag_auv
return mapping
def group_video_segments(video_list: list[dict], gap_min: int = 5) -> list[dict]:
"""Group consecutive videos into segments by timestamp gap."""
sorted_vids = sorted(
[v for v in video_list if v.get("create_date")],
key=lambda x: x["create_date"]
)
if not sorted_vids:
return [{"videos": video_list, "label": "seg_unknown", "total_s": None}]
segments = []
current: list[dict] = [sorted_vids[0]]
for vid in sorted_vids[1:]:
prev = current[-1]
try:
prev_end = datetime.fromisoformat(prev["create_date"])
if prev.get("duration_s"):
prev_end += timedelta(seconds=prev["duration_s"])
cur_start = datetime.fromisoformat(vid["create_date"])
gap = (cur_start - prev_end).total_seconds() / 60
if gap > gap_min:
segments.append(_finalize_segment(current))
current = [vid]
else:
current.append(vid)
except Exception:
current.append(vid)
if current:
segments.append(_finalize_segment(current))
return segments
def _finalize_segment(videos: list[dict]) -> dict:
label = videos[0]["create_date"][:19].replace(":", "-").replace("T", "_") if videos[0].get("create_date") else "seg_unknown"
total_s = sum(v["duration_s"] or 0 for v in videos)
return {
"label": label,
"videos": videos,
"total_s": total_s,
"start": videos[0].get("create_date"),
"end": videos[-1].get("create_date"),
}
def build_manifest(mission_path: Path, gap_min: int = 5) -> dict:
raw_data = mission_path / "raw_data"
if not raw_data.exists():
# try direct
raw_data = mission_path
print(f"[01_ingest] scanning {raw_data} ...")
videos = scan_videos(raw_data)
bags = scan_bags(raw_data)
csvs = scan_usbl_csv(raw_data)
auv_map = detect_auv_mapping(videos, bags)
# Build per-AUV segments
auv_segments: dict[str, list[dict]] = {}
for auv_id, vid_list in videos.items():
segs = group_video_segments(vid_list, gap_min=gap_min)
auv_segments[auv_id] = segs
# Compute AUVs with real data
auv_ids_with_video = sorted(videos.keys())
auv_ids_with_bags = sorted(bags.keys())
total_video_s = sum(
seg["total_s"] or 0
for segs in auv_segments.values()
for seg in segs
)
manifest = {
"mission": mission_path.name,
"ssd_path": str(mission_path),
"generated_at": now_iso(),
"auv_ids_video": auv_ids_with_video,
"auv_ids_bags": auv_ids_with_bags,
"auv_mapping": auv_map,
"total_video_s": round(total_video_s),
"segments_per_auv": auv_segments,
"bag_sessions_per_auv": bags,
"usbl_csv_per_auv": csvs,
"notes": {
"usbl_csv_format": "raw_serial_bytes",
"nav_source": "mcap_bags",
},
}
return manifest
def ingest(mission_path: Path, mission_name: str, out_dir: Path,
gap_min: int = 5) -> dict:
out_dir.mkdir(parents=True, exist_ok=True)
manifest_path = out_dir / mission_name / "manifest.json"
# Idempotency check
if manifest_path.exists():
existing = json.loads(manifest_path.read_text())
chk = hashlib.sha256(mission_path.name.encode()).hexdigest()[:8]
print(f"[01_ingest] manifest exists (checksum {chk}), skipping scan")
return existing
manifest = build_manifest(mission_path, gap_min=gap_min)
# Save manifest
manifest_path.parent.mkdir(parents=True, exist_ok=True)
manifest_path.write_text(json.dumps(manifest, indent=2))
print(f"[01_ingest] manifest saved: {manifest_path}")
# Write to DB
init_db()
with get_conn() as conn:
upsert_mission(conn, mission_name, str(mission_path),
status="ingested", manifest=json.dumps(manifest))
return manifest
def main():
ap = argparse.ArgumentParser(description="Stage 01 — Ingest mission from SSD")
ap.add_argument("mission_path", type=Path, help="Path to mission folder (e.g. /mnt/ssd/20260505-Lepradet)")
ap.add_argument("--name", type=str, default=None, help="Mission name (defaults to folder name)")
ap.add_argument("--out", type=Path, default=Path("/home/cosma/cosma-pipeline"), help="Output base dir")
ap.add_argument("--gap-min", type=int, default=5, help="Gap in minutes to split video segments")
args = ap.parse_args()
mission_name = args.name or args.mission_path.name
manifest = ingest(args.mission_path, mission_name, args.out, args.gap_min)
print(f"\n=== Ingest summary for {mission_name} ===")
print(f"AUVs with video: {manifest['auv_ids_video']}")
print(f"AUVs with bags: {manifest['auv_ids_bags']}")
print(f"AUV mapping: {manifest['auv_mapping']}")
print(f"Total video: {manifest['total_video_s']}s")
print(f"Segments:")
for auv, segs in manifest["segments_per_auv"].items():
for seg in segs:
n_vids = len(seg["videos"])
dur = f"{seg['total_s']:.0f}s" if seg["total_s"] else "?"
print(f" {auv} / {seg['label']} {n_vids} videos {dur}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,272 @@
#!/usr/bin/env python3
"""Stage 02 — Parse navigation from ROS2 MCAP bag files.
Extracts per-AUV trajectories from MCAP bags using mcap_ros2:
- /mavros/global_position/global → NavSatFix (lat, lon, alt)
- /mavros/imu/data → Imu (qx, qy, qz, qw)
- /mavros/imu/static_pressure → FluidPressure (pressure_pa)
Joins on nearest timestamp (tolerance 100ms).
Saves parquet: ~/cosma-pipeline/data/<mission>/nav/<AUV>_<segment>.parquet
Fallback: if no MCAP GPS data, marks as degraded=True (GPS=0 under water is normal).
Usage:
python3 02_nav_parse.py /home/cosma/cosma-pipeline/20260505-Lepradet/manifest.json
python3 02_nav_parse.py /path/manifest.json --auv AUV013
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).parent.parent))
from orchestrator.db import init_db, get_conn, upsert_job, record_metric, now_iso
PIPELINE_BASE = Path(os.environ.get("COSMA_PIPELINE_BASE", "/home/cosma/cosma-pipeline"))
NAV_TOPICS = [
"/mavros/global_position/global",
"/mavros/imu/data",
"/mavros/imu/static_pressure",
]
def parse_mcap_segment(mcap_files: list[Path]) -> dict[str, list]:
"""Extract raw topic data from a list of MCAP files (one session/segment).
Returns dict keyed by topic -> list of (ts_ns, data_dict).
"""
from mcap_ros2.reader import read_ros2_messages
topic_data: dict[str, list] = {t: [] for t in NAV_TOPICS}
for mcap_path in mcap_files:
if not mcap_path.exists():
continue
try:
for msg in read_ros2_messages(str(mcap_path), topics=NAV_TOPICS):
topic = msg.channel.topic
m = msg.ros_msg
ts_ns = int(msg.log_time.timestamp() * 1e9)
if topic == "/mavros/global_position/global":
topic_data[topic].append((ts_ns, {
"lat": float(m.latitude),
"lon": float(m.longitude),
"alt": float(m.altitude),
}))
elif topic == "/mavros/imu/data":
topic_data[topic].append((ts_ns, {
"qx": float(m.orientation.x),
"qy": float(m.orientation.y),
"qz": float(m.orientation.z),
"qw": float(m.orientation.w),
}))
elif topic == "/mavros/imu/static_pressure":
topic_data[topic].append((ts_ns, {
"pressure_pa": float(m.fluid_pressure),
}))
except Exception as e:
print(f" [02] Error reading {mcap_path.name}: {e}")
return topic_data
def join_topics(topic_data: dict[str, list], tol_ns: int = 100_000_000) -> list[dict]:
"""Join NavSatFix + Imu + FluidPressure on nearest timestamp (100ms tol).
Base timeline = NavSatFix if available, else Imu.
"""
import pandas as pd
nav_pts = topic_data.get("/mavros/global_position/global", [])
imu_pts = topic_data.get("/mavros/imu/data", [])
pres_pts = topic_data.get("/mavros/imu/static_pressure", [])
if not nav_pts and not imu_pts:
return []
# Build DataFrames
if nav_pts:
df_nav = pd.DataFrame([{"ts_ns": ts, **d} for ts, d in nav_pts])
else:
df_nav = pd.DataFrame(columns=["ts_ns", "lat", "lon", "alt"])
if imu_pts:
df_imu = pd.DataFrame([{"ts_ns": ts, **d} for ts, d in imu_pts])
else:
df_imu = pd.DataFrame(columns=["ts_ns", "qx", "qy", "qz", "qw"])
if pres_pts:
df_pres = pd.DataFrame([{"ts_ns": ts, **d} for ts, d in pres_pts])
else:
df_pres = pd.DataFrame(columns=["ts_ns", "pressure_pa"])
# Use nav as base if it has data, else imu
base_df = df_nav if len(df_nav) > 0 else df_imu
base_df = base_df.sort_values("ts_ns").reset_index(drop=True)
# Merge-as-of for IMU
result = base_df.copy()
if len(df_imu) > 0:
df_imu_s = df_imu.sort_values("ts_ns").reset_index(drop=True)
# Simple nearest-neighbor join
imu_ts = df_imu_s["ts_ns"].values
for col in ["qx", "qy", "qz", "qw"]:
result[col] = np.nan
for i, row_ts in enumerate(result["ts_ns"].values):
idx = np.argmin(np.abs(imu_ts - row_ts))
if abs(imu_ts[idx] - row_ts) <= tol_ns:
for col in ["qx", "qy", "qz", "qw"]:
result.at[i, col] = float(df_imu_s.at[idx, col])
# Merge pressure
if len(df_pres) > 0:
df_pres_s = df_pres.sort_values("ts_ns").reset_index(drop=True)
pres_ts = df_pres_s["ts_ns"].values
result["pressure_pa"] = np.nan
for i, row_ts in enumerate(result["ts_ns"].values):
idx = np.argmin(np.abs(pres_ts - row_ts))
if abs(pres_ts[idx] - row_ts) <= tol_ns:
result.at[i, "pressure_pa"] = float(df_pres_s.at[idx, "pressure_pa"])
# Ensure all columns exist
for col in ["lat", "lon", "alt", "qx", "qy", "qz", "qw", "pressure_pa"]:
if col not in result.columns:
result[col] = np.nan
return result.to_dict("records")
def parse_auv(manifest: dict, auv_id: str, out_dir: Path) -> dict:
"""Parse all MCAP sessions for one AUV. Returns metrics."""
from pathlib import Path as P
metrics = {
"auv_id": auv_id,
"segments": [],
"total_points": 0,
"degraded": False,
"status": "ok",
}
bag_sessions = manifest.get("bag_sessions_per_auv", {}).get(auv_id, [])
if not bag_sessions:
auv_map = manifest.get("auv_mapping", {})
bag_auv = auv_map.get(auv_id)
if bag_auv:
bag_sessions = manifest.get("bag_sessions_per_auv", {}).get(bag_auv, [])
if not bag_sessions:
# Build from raw SSD structure
ssd_path = P(manifest.get("ssd_path", "/mnt/ssd") + "/" + manifest["mission"].split("-")[0] + "-" + manifest["mission"].split("-")[1] if "-" in manifest["mission"] else manifest.get("ssd_path", "/mnt/ssd"))
auv_num = auv_id.replace("AUV", "0") # AUV013 -> 0013? No: AUV013 -> AUV013
bag_root = P(manifest.get("ssd_path", "/mnt/ssd")) / "raw_data/logs/SUB/bag"
sessions = sorted(bag_root.glob(f"*_{auv_id}"))
bag_sessions = [{"label": s.name, "mcap_files": [str(f) for f in sorted(s.glob("*.mcap"))]} for s in sessions]
import pandas as pd
all_points_total = 0
for sess in bag_sessions:
label = sess.get("session", sess.get("label", "unknown"))
mcap_files = [P(f) for f in sess.get("mcap_files", [])]
if not mcap_files:
continue
out_parquet = out_dir / f"{auv_id}_{label}.parquet"
if out_parquet.exists():
df_ex = pd.read_parquet(out_parquet)
n = len(df_ex)
print(f" [02] {auv_id}/{label}: cached ({n} pts)")
all_points_total += n
metrics["segments"].append({"label": label, "points": n, "cached": True})
continue
print(f" [02] {auv_id}/{label}: parsing {len(mcap_files)} MCAP files...")
topic_data = parse_mcap_segment(mcap_files)
points = join_topics(topic_data)
if not points:
print(f" [02] {auv_id}/{label}: no data")
metrics["segments"].append({"label": label, "points": 0, "degraded": True})
metrics["degraded"] = True
continue
df = pd.DataFrame(points)
n = len(df)
# Check GPS quality
has_gps = df["lat"].notna().any() and (df["lat"] != 0).any()
if not has_gps:
print(f" [02] {auv_id}/{label}: {n} pts, GPS=0 (degraded — AUV underwater)")
metrics["degraded"] = True
else:
print(f" [02] {auv_id}/{label}: {n} pts, GPS OK")
df.to_parquet(out_parquet, index=False)
all_points_total += n
metrics["segments"].append({"label": label, "points": n, "degraded": not has_gps})
metrics["total_points"] = all_points_total
if all_points_total == 0:
metrics["status"] = "degraded"
return metrics
def parse_mission(manifest_path: Path, auv_filter: str | None = None) -> list[dict]:
manifest = json.loads(manifest_path.read_text())
mission_name = manifest["mission"]
out_dir = PIPELINE_BASE / "data" / mission_name / "nav"
out_dir.mkdir(parents=True, exist_ok=True)
auv_ids = list(set(
manifest.get("auv_ids_bags", []) +
list(manifest.get("auv_mapping", {}).keys())
))
if not auv_ids:
auv_ids = manifest.get("auv_ids_video", [])
if auv_filter:
auv_ids = [a for a in auv_ids if a == auv_filter]
all_metrics = []
init_db()
for auv_id in sorted(auv_ids):
print(f"[02] === {auv_id} ===")
m = parse_auv(manifest, auv_id, out_dir)
all_metrics.append(m)
with get_conn() as conn:
mission_row = conn.execute("SELECT id FROM missions WHERE name=?", (mission_name,)).fetchone()
if mission_row:
job_id = upsert_job(conn, mission_row["id"], auv_id, "all", "02_nav_parse",
status="done" if m["status"] == "ok" else m["status"],
output_path=str(out_dir))
record_metric(conn, job_id, "nav_points_total", value=m["total_points"],
pass_fail="pass" if m["total_points"] > 0 else "warn")
return all_metrics
def main():
ap = argparse.ArgumentParser(description="Stage 02 — Parse nav from MCAP bags")
ap.add_argument("manifest", type=Path)
ap.add_argument("--auv", type=str, default=None)
args = ap.parse_args()
metrics = parse_mission(args.manifest, auv_filter=args.auv)
print("\n=== Stage 02 summary ===")
for m in metrics:
segs = m.get("segments", [])
total = m.get("total_points", 0)
deg = "DEGRADED" if m.get("degraded") else "OK"
print(f" {m['auv_id']}: {total} pts across {len(segs)} segments [{deg}]")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,272 @@
#!/usr/bin/env python3
"""Stage 02 — Parse USBL/navigation from MCAP bag files.
The USBL CSV logs in logs/SUB/log/ contain raw serial bytes, NOT lat/lon.
Real navigation data is in MCAP bags (logs/SUB/bag/).
This stage:
1. Reads MCAP files per AUV session
2. Extracts position topics (configurable in default_params.yaml)
3. Falls back to parsing serial bytes if no nav topic found (best-effort)
4. Outputs Parquet per AUV with columns: timestamp, lat, lon, depth, heading
Usage:
python3 02_usbl_parse.py /home/cosma/cosma-pipeline/20260505-Lepradet/manifest.json
python3 02_usbl_parse.py /home/cosma/cosma-pipeline/20260505-Lepradet/manifest.json --auv AUV010
"""
from __future__ import annotations
import argparse
import json
import os
import struct
import sys
from datetime import datetime, timezone
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).parent.parent))
from orchestrator.db import init_db, get_conn, upsert_job, record_metric, now_iso
LOG_DIR = Path(os.environ.get("COSMA_PIPELINE_LOGS", "/home/cosma/cosma-pipeline/logs"))
LOG_DIR.mkdir(parents=True, exist_ok=True)
# Known nav topics to try in MCAP files
NAV_TOPICS = [
"/usbl/position",
"/usbl/fix",
"/navigation/position",
"/bluerov/usbl",
"/waterlinked/position",
"/fix",
"/gps/fix",
"/mavros/global_position/global",
"/mavros/global_position/local",
]
def try_parse_mcap(mcap_path: Path, topics: list[str] | None = None) -> list[dict]:
"""Try to extract nav points from MCAP file. Returns list of {ts, lat, lon, depth}."""
try:
from mcap.reader import make_reader
except ImportError:
print(f" [02] mcap not installed, skipping {mcap_path.name}")
return []
points = []
try:
with open(mcap_path, "rb") as f:
reader = make_reader(f)
for schema, channel, message in reader.iter_messages(topics=topics):
# Try to deserialize — support ROS2 JSON-encoded or raw
try:
import json as _json
data = _json.loads(message.data)
lat = data.get("latitude") or data.get("lat")
lon = data.get("longitude") or data.get("lon")
depth = data.get("depth") or data.get("altitude")
heading = data.get("heading") or data.get("yaw")
if lat is not None and lon is not None:
ts_ns = message.log_time
ts = ts_ns / 1e9
points.append({
"timestamp": ts,
"lat": float(lat),
"lon": float(lon),
"depth": float(depth) if depth is not None else None,
"heading": float(heading) if heading is not None else None,
"source": channel.topic,
})
except Exception:
pass
except Exception as e:
print(f" [02] MCAP read error {mcap_path.name}: {e}")
return points
def try_parse_serial_csv(csv_path: Path) -> list[dict]:
"""Best-effort: parse raw serial byte log for USBL range/bearing frames.
Waterlinked USBL M64 protocol: 0xBB 0x55 frame header.
This is a rough attempt — actual lat/lon requires ship GPS + range+bearing.
Returns relative-only positions (range, bearing) if decoded.
"""
points = []
try:
with open(csv_path, "r") as f:
for line in f:
line = line.strip()
if not line:
continue
parts = line.split(",", 2)
if len(parts) < 3:
continue
ts_str, direction, raw = parts[0], parts[1], parts[2]
if direction.strip() != "RECEIVED":
continue
# Extract bytes from repr string b'\xbb...'
try:
raw_clean = raw.strip().strip('"')
# Parse Python bytes repr
data = eval(raw_clean) # safe: only used on known CSV
if len(data) >= 4 and data[0] == 0xBB and data[1] == 0x55:
# Waterlinked M64 position frame: len byte at [2], payload follows
payload_len = data[2]
if len(data) >= payload_len + 3:
# Best effort: look for float32 values in payload
# Actual protocol decoding would need WL M64 spec
ts = datetime.fromisoformat(ts_str)
points.append({
"timestamp": ts.timestamp(),
"lat": None,
"lon": None,
"depth": None,
"heading": None,
"source": "serial_raw",
"raw_bytes": data.hex()[:32],
})
except Exception:
pass
except Exception as e:
print(f" [02] serial CSV parse error {csv_path.name}: {e}")
return points
def parse_auv_sessions(manifest: dict, auv_id: str, out_dir: Path,
topics: list[str] | None = None) -> dict:
"""Parse all sessions for one AUV. Returns metrics dict."""
metrics = {"auv_id": auv_id, "points_raw": 0, "sources": [], "status": "ok"}
all_points: list[dict] = []
# Try MCAP bags first
bag_sessions = manifest.get("bag_sessions_per_auv", {}).get(auv_id, [])
if not bag_sessions:
# Try mapping: maybe bags use AUV0xx while videos use AUV2xx
auv_map = manifest.get("auv_mapping", {})
bag_auv = auv_map.get(auv_id)
if bag_auv:
bag_sessions = manifest.get("bag_sessions_per_auv", {}).get(bag_auv, [])
for sess in bag_sessions:
for mcap_path_str in sess.get("mcap_files", []):
mcap_path = Path(mcap_path_str)
if not mcap_path.exists():
continue
pts = try_parse_mcap(mcap_path, topics=topics or NAV_TOPICS)
if pts:
all_points.extend(pts)
if "mcap" not in metrics["sources"]:
metrics["sources"].append("mcap")
print(f" [02] {auv_id} {mcap_path.name}: {len(pts)} nav points")
# Fallback: serial CSV if no MCAP nav
if not all_points:
csv_entries = manifest.get("usbl_csv_per_auv", {}).get(auv_id, [])
for entry in csv_entries:
csv_path = Path(entry["path"])
if not csv_path.exists():
continue
pts = try_parse_serial_csv(csv_path)
if pts:
all_points.extend(pts)
if "serial_csv" not in metrics["sources"]:
metrics["sources"].append("serial_csv")
metrics["points_raw"] = len(all_points)
if not all_points:
metrics["status"] = "degraded"
metrics["note"] = "no nav points found in MCAP or serial CSV"
print(f" [02] {auv_id}: NO nav data found — degraded")
else:
print(f" [02] {auv_id}: {len(all_points)} raw nav points from {metrics['sources']}")
# Save output
out_file = out_dir / f"{auv_id}_nav_raw.json"
out_file.write_text(json.dumps({
"auv_id": auv_id,
"generated_at": now_iso(),
"metrics": metrics,
"points": all_points,
}, indent=2, default=str))
# Also save as simple CSV for downstream
csv_out = out_dir / f"{auv_id}_nav_raw.csv"
with open(csv_out, "w") as f:
f.write("timestamp,lat,lon,depth,heading,source\n")
for p in all_points:
f.write(f"{p['timestamp']},{p.get('lat','')},{p.get('lon','')},{p.get('depth','')},{p.get('heading','')},{p.get('source','')}\n")
return metrics
def parse_mission(manifest_path: Path, auv_filter: str | None = None,
out_dir: Path | None = None) -> list[dict]:
manifest = json.loads(manifest_path.read_text())
mission_name = manifest["mission"]
if out_dir is None:
out_dir = manifest_path.parent / "02_usbl_raw"
out_dir.mkdir(parents=True, exist_ok=True)
# Idempotency: check if all AUV outputs exist
auv_ids = manifest.get("auv_ids_bags", []) or manifest.get("auv_ids_video", [])
if auv_filter:
auv_ids = [a for a in auv_ids if a == auv_filter]
all_metrics = []
init_db()
for auv_id in auv_ids:
out_file = out_dir / f"{auv_id}_nav_raw.json"
if out_file.exists():
print(f"[02] {auv_id}: output exists, skipping")
existing = json.loads(out_file.read_text())
all_metrics.append(existing.get("metrics", {"auv_id": auv_id, "status": "cached"}))
continue
print(f"[02] Parsing {auv_id} ...")
m = parse_auv_sessions(manifest, auv_id, out_dir)
all_metrics.append(m)
# Record in DB
with get_conn() as conn:
from orchestrator.db import upsert_mission
mission_id_row = conn.execute(
"SELECT id FROM missions WHERE name=?", (mission_name,)
).fetchone()
if mission_id_row:
mission_id = mission_id_row["id"]
job_id = upsert_job(conn, mission_id, auv_id, "all", "02_usbl_parse",
status="done" if m["status"] == "ok" else m["status"],
output_path=str(out_dir))
record_metric(conn, job_id, "usbl_points_raw",
value=m["points_raw"],
pass_fail="pass" if m["points_raw"] > 0 else "fail")
return all_metrics
def main():
ap = argparse.ArgumentParser(description="Stage 02 — Parse USBL/nav from MCAP bags")
ap.add_argument("manifest", type=Path, help="manifest.json from stage 01")
ap.add_argument("--auv", type=str, default=None, help="Filter to single AUV ID")
ap.add_argument("--out", type=Path, default=None, help="Output directory")
args = ap.parse_args()
metrics = parse_mission(args.manifest, auv_filter=args.auv, out_dir=args.out)
print("\n=== Stage 02 summary ===")
total_pts = sum(m.get("points_raw", 0) for m in metrics)
for m in metrics:
status = m.get("status", "?")
pts = m.get("points_raw", 0)
src = m.get("sources", [])
print(f" {m['auv_id']}: {pts} pts {src} [{status}]")
print(f"Total nav points: {total_pts}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,185 @@
#!/usr/bin/env python3
"""Stage 03 — Filter and smooth navigation trajectories.
Input: ~/cosma-pipeline/data/<mission>/nav/<AUV>_<segment>.parquet
Output: ~/cosma-pipeline/data/<mission>/nav_filtered/<AUV>_<segment>.parquet
Steps:
1. Drop rows with null lat/lon OR lat==0 AND lon==0 (no GPS lock)
2. MAD-3σ outlier removal on lat, lon
3. Moving average smoothing (window 5s, KISS)
4. Depth from pressure: depth_m = (pressure_pa - 101325) / (1025 * 9.81)
5. Output: same columns + lat_smooth, lon_smooth, depth_m
Usage:
python3 03_nav_filter.py /home/cosma/cosma-pipeline/data/20260505-Lepradet/nav/
python3 03_nav_filter.py /path/nav/ --auv AUV013 --sigma 2.5
"""
from __future__ import annotations
import argparse
import os
import sys
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).parent.parent))
from orchestrator.db import init_db, get_conn, upsert_job, record_metric, now_iso
PIPELINE_BASE = Path(os.environ.get("COSMA_PIPELINE_BASE", "/home/cosma/cosma-pipeline"))
RHO_SEA = 1025.0 # kg/m3
G = 9.81 # m/s2
P_ATM = 101325.0 # Pa
def mad_mask(arr: np.ndarray, sigma: float = 3.0) -> np.ndarray:
"""True = keep."""
if len(arr) < 4:
return np.ones(len(arr), dtype=bool)
med = np.median(arr)
mad = np.median(np.abs(arr - med))
if mad == 0:
return np.ones(len(arr), dtype=bool)
return np.abs(0.6745 * (arr - med) / mad) < sigma
def moving_average(arr: np.ndarray, window: int = 5) -> np.ndarray:
if len(arr) < window:
return arr.copy()
pad = window // 2
padded = np.pad(arr, (pad, pad), mode="edge")
return np.convolve(padded, np.ones(window) / window, mode="valid")[:len(arr)]
def filter_parquet(src: Path, dst_dir: Path, sigma: float = 3.0, window: int = 5) -> dict:
import pandas as pd
df = pd.read_parquet(src)
auv_seg = src.stem
metrics = {
"file": src.name,
"points_in": len(df),
"points_out": 0,
"status": "ok",
}
# Step 1: drop null/zero GPS
has_lat = "lat" in df.columns and df["lat"].notna().any()
if has_lat:
mask_valid = df["lat"].notna() & df["lon"].notna() & (df["lat"] != 0) & (df["lon"] != 0)
df_valid = df[mask_valid].copy()
else:
# No GPS — keep all rows for IMU/pressure
df_valid = df.copy()
metrics["degraded"] = True
if len(df_valid) == 0:
metrics["status"] = "degraded"
metrics["note"] = "no valid GPS points"
print(f" [03] {auv_seg}: no valid GPS — saving as-is with depth calc only")
df_out = df.copy()
else:
# Step 2: MAD outlier removal on lat/lon
if has_lat and len(df_valid) >= 4:
lats = df_valid["lat"].values
lons = df_valid["lon"].values
mask = mad_mask(lats, sigma) & mad_mask(lons, sigma)
n_removed = int((~mask).sum())
df_valid = df_valid[mask].copy()
metrics["points_removed_outlier"] = n_removed
else:
metrics["points_removed_outlier"] = 0
# Step 3: sort by timestamp
if "ts_ns" in df_valid.columns:
df_valid = df_valid.sort_values("ts_ns").reset_index(drop=True)
# Step 4: smooth lat/lon
if has_lat and len(df_valid) >= window:
df_valid["lat_smooth"] = moving_average(df_valid["lat"].values, window)
df_valid["lon_smooth"] = moving_average(df_valid["lon"].values, window)
elif has_lat and len(df_valid) > 0:
df_valid["lat_smooth"] = df_valid["lat"]
df_valid["lon_smooth"] = df_valid["lon"]
else:
df_valid["lat_smooth"] = np.nan
df_valid["lon_smooth"] = np.nan
df_out = df_valid
# Step 5: depth from pressure
if "pressure_pa" in df_out.columns and df_out["pressure_pa"].notna().any():
df_out["depth_m"] = (df_out["pressure_pa"] - P_ATM) / (RHO_SEA * G)
df_out["depth_m"] = df_out["depth_m"].abs() # negative when underwater (P < Patm) # surface = 0
else:
df_out["depth_m"] = np.nan
dst_dir.mkdir(parents=True, exist_ok=True)
out_path = dst_dir / src.name
df_out.to_parquet(out_path, index=False)
metrics["points_out"] = len(df_out)
removed_null = metrics["points_in"] - len(df_out) - metrics.get("points_removed_outlier", 0)
metrics["points_removed_null"] = max(0, removed_null)
print(f" [03] {auv_seg}: {metrics['points_in']}{metrics['points_out']} pts, "
f"depth_m range=[{df_out['depth_m'].min():.1f}, {df_out['depth_m'].max():.1f}]"
if df_out["depth_m"].notna().any() else
f" [03] {auv_seg}: {metrics['points_in']}{metrics['points_out']} pts, no pressure")
return metrics
def filter_mission(nav_dir: Path, auv_filter: str | None = None,
sigma: float = 3.0, window: int = 5) -> list[dict]:
out_dir = nav_dir.parent / "nav_filtered"
parquet_files = sorted(nav_dir.glob("*.parquet"))
if auv_filter:
parquet_files = [f for f in parquet_files if auv_filter in f.name]
all_metrics = []
init_db()
for pf in parquet_files:
out_file = out_dir / pf.name
if out_file.exists():
print(f"[03] {pf.stem}: cached")
continue
print(f"[03] Filtering {pf.name}...")
m = filter_parquet(pf, out_dir, sigma=sigma, window=window)
all_metrics.append(m)
with get_conn() as conn:
mission_name = nav_dir.parent.name
mission_row = conn.execute("SELECT id FROM missions WHERE name=?", (mission_name,)).fetchone()
if mission_row:
auv_id = pf.stem.split("_")[0]
job_id = upsert_job(conn, mission_row["id"], auv_id, "all", "03_nav_filter",
status="done" if m.get("status") == "ok" else m.get("status", "done"),
output_path=str(out_dir))
record_metric(conn, job_id, "nav_points_filtered", value=m.get("points_out", 0),
pass_fail="pass" if m.get("points_out", 0) > 0 else "warn")
return all_metrics
def main():
ap = argparse.ArgumentParser(description="Stage 03 — Filter nav trajectories")
ap.add_argument("nav_dir", type=Path, help="Directory with *.parquet from stage 02")
ap.add_argument("--auv", type=str, default=None)
ap.add_argument("--sigma", type=float, default=3.0)
ap.add_argument("--window", type=int, default=5)
args = ap.parse_args()
metrics = filter_mission(args.nav_dir, auv_filter=args.auv,
sigma=args.sigma, window=args.window)
print("\n=== Stage 03 summary ===")
for m in metrics:
print(f" {m.get('file','?')}: {m.get('points_in',0)}{m.get('points_out',0)} "
f"[{m.get('status','?')}]")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,248 @@
#!/usr/bin/env python3
"""Stage 03 — Filter and smooth USBL navigation trajectory.
Input: 02_usbl_raw/<AUV>_nav_raw.json (or .csv)
Output: 03_usbl_filtered/<AUV>_nav_filtered.json + .csv
Steps:
1. Drop points with null lat/lon
2. MAD-3σ outlier removal on lat, lon, depth independently
3. Moving-average smoothing (window=5 by default)
4. Optional: simple 1D Kalman on each axis (KISS — no cross-covariance)
Usage:
python3 03_usbl_filter.py /home/cosma/cosma-pipeline/20260505-Lepradet/02_usbl_raw/
python3 03_usbl_filter.py /path/to/02_usbl_raw/ --auv AUV010 --sigma 2.5
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).parent.parent))
from orchestrator.db import init_db, get_conn, upsert_job, record_metric, now_iso
LOG_DIR = Path(os.environ.get("COSMA_PIPELINE_LOGS", "/home/cosma/cosma-pipeline/logs"))
LOG_DIR.mkdir(parents=True, exist_ok=True)
def mad_outlier_mask(arr: np.ndarray, sigma: float = 3.0) -> np.ndarray:
"""Returns boolean mask: True = keep (inlier). Uses MAD-based sigma."""
if len(arr) < 4:
return np.ones(len(arr), dtype=bool)
median = np.median(arr)
mad = np.median(np.abs(arr - median))
if mad == 0:
return np.ones(len(arr), dtype=bool)
modified_z = 0.6745 * (arr - median) / mad
return np.abs(modified_z) < sigma
def moving_average(arr: np.ndarray, window: int = 5) -> np.ndarray:
"""Centered moving average with edge padding."""
if len(arr) < window:
return arr.copy()
pad = window // 2
padded = np.pad(arr, (pad, pad), mode="edge")
kernel = np.ones(window) / window
return np.convolve(padded, kernel, mode="valid")[:len(arr)]
def simple_kalman_1d(measurements: np.ndarray,
process_noise: float = 1e-4,
measurement_noise: float = 1e-2) -> np.ndarray:
"""Very simple 1D Kalman filter (scalar, no velocity state).
KISS: just smooths, no cross-axis coupling.
"""
n = len(measurements)
filtered = np.zeros(n)
x_est = measurements[0]
p_est = 1.0
for i, z in enumerate(measurements):
# Predict
p_pred = p_est + process_noise
# Update
K = p_pred / (p_pred + measurement_noise)
x_est = x_est + K * (z - x_est)
p_est = (1 - K) * p_pred
filtered[i] = x_est
return filtered
def filter_auv_nav(raw_path: Path, out_path: Path,
sigma: float = 3.0, window: int = 5,
use_kalman: bool = False) -> dict:
"""Filter nav for one AUV. Returns metrics dict."""
data = json.loads(raw_path.read_text())
points = data.get("points", [])
auv_id = data.get("auv_id", raw_path.stem.replace("_nav_raw", ""))
metrics = {
"auv_id": auv_id,
"points_before": len(points),
"points_after": 0,
"points_removed_null": 0,
"points_removed_outlier": 0,
"status": "ok",
}
if not points:
metrics["status"] = "degraded"
metrics["note"] = "no points to filter"
_save_output(auv_id, [], out_path, metrics)
return metrics
# Step 1: Drop null lat/lon
valid = [p for p in points if p.get("lat") is not None and p.get("lon") is not None]
metrics["points_removed_null"] = len(points) - len(valid)
if not valid:
metrics["status"] = "degraded"
metrics["note"] = "all points have null lat/lon (serial-only data)"
print(f" [03] {auv_id}: all null lat/lon — degraded (serial CSV source, no MCAP nav)")
_save_output(auv_id, [], out_path, metrics)
return metrics
# Step 2: MAD outlier removal
lats = np.array([p["lat"] for p in valid])
lons = np.array([p["lon"] for p in valid])
mask_lat = mad_outlier_mask(lats, sigma)
mask_lon = mad_outlier_mask(lons, sigma)
mask = mask_lat & mask_lon
# Also filter depth if present
depths = np.array([p.get("depth") or np.nan for p in valid])
if not np.all(np.isnan(depths)):
mask_depth = mad_outlier_mask(depths[~np.isnan(depths)], sigma)
# Map back — only filter where we have depth
depth_idx = np.where(~np.isnan(depths))[0]
for i, keep in zip(depth_idx, mask_depth):
if not keep:
mask[i] = False
filtered_points = [p for p, keep in zip(valid, mask) if keep]
metrics["points_removed_outlier"] = int(np.sum(~mask))
# Step 3: Sort by timestamp
filtered_points.sort(key=lambda p: p["timestamp"])
# Step 4: Smooth
if len(filtered_points) >= window:
filt_lats = moving_average(np.array([p["lat"] for p in filtered_points]), window)
filt_lons = moving_average(np.array([p["lon"] for p in filtered_points]), window)
for i, p in enumerate(filtered_points):
p = dict(p)
p["lat"] = float(filt_lats[i])
p["lon"] = float(filt_lons[i])
filtered_points[i] = p
# Step 5: Optional Kalman
if use_kalman and len(filtered_points) > 4:
k_lats = simple_kalman_1d(np.array([p["lat"] for p in filtered_points]))
k_lons = simple_kalman_1d(np.array([p["lon"] for p in filtered_points]))
for i, p in enumerate(filtered_points):
p = dict(p)
p["lat"] = float(k_lats[i])
p["lon"] = float(k_lons[i])
filtered_points[i] = p
metrics["points_after"] = len(filtered_points)
if metrics["points_after"] < 5:
metrics["status"] = "degraded"
metrics["note"] = f"too few points after filter: {metrics['points_after']}"
print(f" [03] {auv_id}: {metrics['points_before']}{metrics['points_after']} "
f"(removed {metrics['points_removed_null']} null, {metrics['points_removed_outlier']} outliers)")
_save_output(auv_id, filtered_points, out_path, metrics)
return metrics
def _save_output(auv_id: str, points: list[dict], out_dir: Path, metrics: dict) -> None:
out_dir.mkdir(parents=True, exist_ok=True)
json_out = out_dir / f"{auv_id}_nav_filtered.json"
json_out.write_text(json.dumps({
"auv_id": auv_id,
"generated_at": now_iso(),
"metrics": metrics,
"points": points,
}, indent=2, default=str))
csv_out = out_dir / f"{auv_id}_nav_filtered.csv"
with open(csv_out, "w") as f:
f.write("timestamp,lat,lon,depth,heading,source\n")
for p in points:
f.write(f"{p['timestamp']},{p.get('lat','')},{p.get('lon','')},{p.get('depth','')},{p.get('heading','')},{p.get('source','')}\n")
def filter_mission(raw_dir: Path, out_dir: Path | None = None,
auv_filter: str | None = None,
sigma: float = 3.0, window: int = 5,
use_kalman: bool = False) -> list[dict]:
if out_dir is None:
out_dir = raw_dir.parent / "03_usbl_filtered"
raw_files = sorted(raw_dir.glob("*_nav_raw.json"))
if auv_filter:
raw_files = [f for f in raw_files if auv_filter in f.name]
all_metrics = []
init_db()
for raw_file in raw_files:
out_file = out_dir / raw_file.name.replace("_raw", "_filtered")
if out_file.exists():
print(f"[03] {raw_file.stem}: output exists, skipping")
existing = json.loads(out_file.read_text())
all_metrics.append(existing.get("metrics", {}))
continue
print(f"[03] Filtering {raw_file.name} ...")
m = filter_auv_nav(raw_file, out_dir, sigma=sigma, window=window, use_kalman=use_kalman)
all_metrics.append(m)
# DB record
with get_conn() as conn:
mission_name = raw_dir.parent.name
mission_row = conn.execute("SELECT id FROM missions WHERE name=?", (mission_name,)).fetchone()
if mission_row:
job_id = upsert_job(conn, mission_row["id"], m["auv_id"], "all", "03_usbl_filter",
status="done" if m["status"] == "ok" else m["status"],
output_path=str(out_dir))
record_metric(conn, job_id, "usbl_points_before", value=m.get("points_before", 0))
record_metric(conn, job_id, "usbl_points_after", value=m.get("points_after", 0),
pass_fail="pass" if m.get("points_after", 0) >= 5 else "fail")
record_metric(conn, job_id, "usbl_points_removed_outlier",
value=m.get("points_removed_outlier", 0))
return all_metrics
def main():
ap = argparse.ArgumentParser(description="Stage 03 — Filter USBL navigation")
ap.add_argument("raw_dir", type=Path, help="Directory with *_nav_raw.json files")
ap.add_argument("--out", type=Path, default=None)
ap.add_argument("--auv", type=str, default=None)
ap.add_argument("--sigma", type=float, default=3.0, help="MAD sigma threshold")
ap.add_argument("--window", type=int, default=5, help="Moving average window")
ap.add_argument("--kalman", action="store_true", help="Apply simple Kalman smoothing")
args = ap.parse_args()
metrics = filter_mission(args.raw_dir, out_dir=args.out, auv_filter=args.auv,
sigma=args.sigma, window=args.window, use_kalman=args.kalman)
print("\n=== Stage 03 summary ===")
for m in metrics:
print(f" {m.get('auv_id','?')}: {m.get('points_before',0)}{m.get('points_after',0)} [{m.get('status','?')}]")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,341 @@
#!/usr/bin/env python3
"""Stage 04 — Extract frames from GoPro videos.
For each MP4 in /mnt/ssd/<mission>/raw_data/medias/videos/GP*-AUV*/
- Skip files < 2MB (placeholders)
- Auto-trim hors-eau: sample frames at start/end, detect non-blue/green pixels
- ffmpeg fps=1, scale=518:294, q:v=3
- Output: ~/cosma-pipeline/data/<mission>/frames/<AUV>/<segment>/frame_XXXXX.jpg
- Skip if output dir exists and has >= expected frames
- Log to SQLite state.db
Usage:
python3 04_frame_extract.py --mission 20260505-Lepradet
python3 04_frame_extract.py --video /mnt/ssd/.../GP1-AUV210/GX019837.MP4
"""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
import time
from pathlib import Path
import cv2
import numpy as np
sys.path.insert(0, str(Path(__file__).parent.parent))
sys.path.insert(0, str(Path(__file__).parent))
from orchestrator.db import init_db, get_conn, upsert_job, record_metric, now_iso
from lib_frame_qc import score_image_file, aggregate as qc_aggregate
QC_SAMPLE_RATE = int(os.environ.get("COSMA_QC_SAMPLE_RATE", "5"))
QC_BOTTOM_OK_PCT = float(os.environ.get("COSMA_QC_BOTTOM_OK_PCT", "50"))
PIPELINE_BASE = Path(os.environ.get("COSMA_PIPELINE_BASE", "/home/cosma/cosma-pipeline"))
SSD_BASE = Path(os.environ.get("COSMA_SSD_BASE", "/mnt/ssd"))
MIN_VIDEO_SIZE_MB = 2.0
def is_underwater_frame(frame_bgr: np.ndarray, threshold: float = 0.6) -> bool:
"""Return True if frame looks like underwater footage (dominant blue/green).
Hors-eau: R > G-5 AND R > B-5 (dry/air dominant).
Underwater: blue or green channel dominant.
"""
b, g, r = cv2.split(frame_bgr.astype(np.float32))
mean_r = float(np.mean(r))
mean_g = float(np.mean(g))
mean_b = float(np.mean(b))
# Not underwater: red dominates
if mean_r > mean_g - 5 and mean_r > mean_b - 5:
return False
return True
def detect_water_range(video_path: Path, sample_count: int = 10) -> tuple[float, float]:
"""Detect start/end times of underwater portion by sampling frames.
Returns (start_s, end_s). Returns (0, duration) if uncertain.
"""
cap = cv2.VideoCapture(str(video_path))
if not cap.isOpened():
return 0.0, -1.0
fps = cap.get(cv2.CAP_PROP_FPS) or 25.0
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
duration_s = total_frames / fps if fps > 0 else 0
# Sample frames: first 20% and last 20%
probe_times_start = [duration_s * i / (sample_count * 5) for i in range(sample_count)]
probe_times_end = [duration_s * (1 - i / (sample_count * 5)) for i in range(sample_count)]
# Find first underwater frame from start
start_s = 0.0
for t in probe_times_start:
cap.set(cv2.CAP_PROP_POS_MSEC, t * 1000)
ret, frame = cap.read()
if ret and is_underwater_frame(frame):
start_s = max(0.0, t - 2.0)
break
# Find last underwater frame from end
end_s = duration_s
for t in sorted(probe_times_end, reverse=True):
cap.set(cv2.CAP_PROP_POS_MSEC, t * 1000)
ret, frame = cap.read()
if ret and is_underwater_frame(frame):
end_s = min(duration_s, t + 2.0)
break
cap.release()
return start_s, end_s
def get_video_duration(video_path: Path) -> float:
"""Get video duration in seconds via ffprobe."""
cmd = [
"ffprobe", "-v", "quiet", "-print_format", "json",
"-show_streams", str(video_path)
]
try:
r = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
data = json.loads(r.stdout)
for stream in data.get("streams", []):
dur = float(stream.get("duration", 0))
if dur > 0:
return dur
except Exception:
pass
return 0.0
def extract_frames(video_path: Path, out_dir: Path, fps: int = 1,
scale: str = "518:294", quality: int = 3,
start_s: float = 0.0, end_s: float = -1.0) -> dict:
"""Run ffmpeg to extract frames. Returns metrics dict."""
out_dir.mkdir(parents=True, exist_ok=True)
# Build ffmpeg args
cmd = ["ffmpeg", "-y", "-loglevel", "error"]
cmd += ["-ss", str(start_s), "-i", str(video_path)]
if end_s > 0 and end_s > start_s:
cmd += ["-t", str(end_s - start_s)]
cmd += [
"-vf", f"fps={fps},scale={scale}",
"-q:v", str(quality),
str(out_dir / "frame_%05d.jpg"),
]
t0 = time.time()
result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600)
elapsed = time.time() - t0
frames = sorted(out_dir.glob("frame_*.jpg"))
n_frames = len(frames)
metrics = {
"video": str(video_path),
"out_dir": str(out_dir),
"n_frames": n_frames,
"elapsed_s": round(elapsed, 1),
"returncode": result.returncode,
"start_s": start_s,
"end_s": end_s,
}
if result.returncode != 0:
metrics["error"] = result.stderr[-500:]
print(f" [04] ffmpeg error for {video_path.name}: {result.stderr[-200:]}")
else:
print(f" [04] {video_path.name}: {n_frames} frames in {elapsed:.1f}s")
# Score a subsample for QC
qc = qc_segment(out_dir, sample_rate=QC_SAMPLE_RATE)
if qc:
metrics.update(qc)
return metrics
def qc_segment(frames_dir: Path, sample_rate: int = 5) -> dict | None:
"""Sample 1/sample_rate frames, score each, write qc.json, return aggregate."""
frames = sorted(frames_dir.glob("frame_*.jpg"))
if not frames:
return None
sampled = frames[::max(1, sample_rate)]
per_frame = []
for f in sampled:
s = score_image_file(f)
if s is not None:
per_frame.append(s)
if not per_frame:
return None
agg = qc_aggregate(per_frame)
qc_payload = {
"frames_in_dir": len(frames),
"frames_sampled": len(per_frame),
"sample_rate": sample_rate,
**agg,
"per_frame": per_frame,
}
try:
(frames_dir / "qc.json").write_text(json.dumps(qc_payload, indent=2))
except Exception as e:
print(f" [04] qc.json write failed: {e}")
print(
f" [04] QC: bottom_visible={agg['bottom_visible_pct']}% "
f"(b={agg['frames_bottom_visible']} ooo={agg['frames_out_of_water']} "
f"turb={agg['frames_turbid']} nob={agg['frames_water_no_bottom']})"
)
return agg
def process_video(video_path: Path, auv_id: str, mission_name: str) -> dict:
"""Process one video file. Returns metrics."""
size_mb = video_path.stat().st_size / (1024 * 1024)
if size_mb < MIN_VIDEO_SIZE_MB:
print(f" [04] Skip {video_path.name} ({size_mb:.1f}MB < {MIN_VIDEO_SIZE_MB}MB)")
return {"video": str(video_path), "skipped": True, "reason": "placeholder"}
segment = video_path.stem
out_dir = PIPELINE_BASE / "data" / mission_name / "frames" / auv_id / segment
# Check if already done
if out_dir.exists():
existing = list(out_dir.glob("frame_*.jpg"))
duration_s = get_video_duration(video_path)
expected = max(1, int(duration_s) - 10) if duration_s > 0 else 1
if len(existing) >= expected:
print(f" [04] {video_path.name}: already done ({len(existing)} frames), skip")
cached_m: dict = {"video": str(video_path), "n_frames": len(existing), "cached": True,
"out_dir": str(out_dir)}
# Re-run QC if qc.json is missing (idempotent enrichment)
if not (out_dir / "qc.json").exists():
qc = qc_segment(out_dir, sample_rate=QC_SAMPLE_RATE)
if qc:
cached_m.update(qc)
else:
try:
cached_qc = json.loads((out_dir / "qc.json").read_text())
for k in (
"frames_total", "frames_bottom_visible", "frames_out_of_water",
"frames_turbid", "frames_water_no_bottom", "bottom_visible_pct",
):
if k in cached_qc:
cached_m[k] = cached_qc[k]
except Exception:
pass
return cached_m
print(f" [04] {video_path.name} ({size_mb:.0f}MB): detecting water range...")
start_s, end_s = detect_water_range(video_path)
print(f" [04] water range: {start_s:.1f}s → {end_s:.1f}s")
return extract_frames(video_path, out_dir, start_s=start_s, end_s=end_s)
def find_auv_videos(mission_path: Path) -> dict[str, list[Path]]:
"""Find all MP4 files per AUV in medias/videos/GP*-AUV*/."""
videos_root = mission_path / "raw_data/medias/videos"
result: dict[str, list[Path]] = {}
for gopro_dir in sorted(videos_root.glob("GP*-AUV*")):
# Extract AUV ID from dir name: GP1-AUV210 -> AUV210
parts = gopro_dir.name.split("-")
if len(parts) >= 2:
auv_id = parts[1]
mp4_files = [f for f in sorted(gopro_dir.glob("GX*.MP4"))
if f.stat().st_size / (1024 * 1024) >= MIN_VIDEO_SIZE_MB]
if mp4_files:
if auv_id not in result:
result[auv_id] = []
result[auv_id].extend(mp4_files)
return result
def process_mission(mission_name: str) -> list[dict]:
mission_path = SSD_BASE / mission_name
auv_videos = find_auv_videos(mission_path)
print(f"[04] Found AUVs: {list(auv_videos.keys())}")
all_metrics = []
init_db()
for auv_id, videos in sorted(auv_videos.items()):
print(f"[04] === {auv_id}: {len(videos)} videos ===")
for video_path in videos:
t0 = time.time()
m = process_video(video_path, auv_id, mission_name)
m["auv_id"] = auv_id
all_metrics.append(m)
if not m.get("skipped"):
with get_conn() as conn:
mission_row = conn.execute(
"SELECT id FROM missions WHERE name=?", (mission_name,)
).fetchone()
if mission_row:
bottom_pct = m.get("bottom_visible_pct")
if m.get("returncode", 0) != 0:
job_status = "error"
elif bottom_pct is not None and bottom_pct < QC_BOTTOM_OK_PCT:
job_status = "degraded"
else:
job_status = "done"
job_id = upsert_job(
conn, mission_row["id"], auv_id,
video_path.stem, "04_frame_extract",
status=job_status,
output_path=m.get("out_dir", ""),
error_msg=(
f"bottom_visible_pct={bottom_pct}% <{QC_BOTTOM_OK_PCT}%"
if job_status == "degraded" else None
),
)
if not m.get("cached"):
record_metric(conn, job_id, "frames_extracted",
value=m.get("n_frames", 0),
pass_fail="pass" if m.get("n_frames", 0) > 0 else "fail")
record_metric(conn, job_id, "extract_time_s",
value=m.get("elapsed_s", 0))
# Always record QC metrics (so cached frames also get scored history)
for k in (
"frames_total", "frames_bottom_visible", "frames_out_of_water",
"frames_turbid", "frames_water_no_bottom",
):
if k in m:
record_metric(conn, job_id, k, value=float(m[k]))
if bottom_pct is not None:
record_metric(
conn, job_id, "bottom_visible_pct",
value=float(bottom_pct),
pass_fail="pass" if bottom_pct >= QC_BOTTOM_OK_PCT else "degraded",
)
return all_metrics
def main():
ap = argparse.ArgumentParser(description="Stage 04 — Extract frames from GoPro videos")
ap.add_argument("--mission", type=str, help="Mission name (e.g. 20260505-Lepradet)")
ap.add_argument("--video", type=Path, help="Single video path")
ap.add_argument("--auv", type=str, default="UNKNOWN", help="AUV ID for single video mode")
args = ap.parse_args()
if args.video:
mission_name = args.mission or "unknown"
m = process_video(args.video, args.auv, mission_name)
print(f"\nResult: {m}")
elif args.mission:
metrics = process_mission(args.mission)
print("\n=== Stage 04 summary ===")
total_frames = sum(m.get("n_frames", 0) for m in metrics if not m.get("skipped"))
skipped = sum(1 for m in metrics if m.get("skipped"))
print(f"Total frames: {total_frames}, skipped: {skipped}")
else:
ap.print_help()
sys.exit(1)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,428 @@
#!/usr/bin/env python3
"""Stage 04b — Trim out-of-water (hors-eau) head/tail frames from already-extracted segments.
Ports the sustained-run trim logic from cosma-qc/scripts/dispatcher.py (_AUTO_TRIM_SCRIPT,
trim_above_water_prefix) into the new cosma-pipeline pipeline. Re-runs frame QC scoring
on the trimmed set and updates state.db (jobs.status + metrics).
Usage:
python3 04b_trim_water.py --mission 20260505-Lepradet
python3 04b_trim_water.py --mission 20260505-Lepradet --auv AUV210 --segment GX019837
python3 04b_trim_water.py --mission 20260505-Lepradet --dry-run
Safety:
- Skips segments where ffmpeg is still running on the frames dir (extraction in progress).
- Skips segments with a queued/running 05_inference job in state.db.
- Skips segments whose frame count is not stable over a 5s window.
- Never deletes all frames (sanity floor: keep everything if trim would empty the dir).
"""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
import time
from pathlib import Path
import cv2
sys.path.insert(0, str(Path(__file__).parent.parent))
sys.path.insert(0, str(Path(__file__).parent))
from orchestrator.db import init_db, get_conn, upsert_job, record_metric, now_iso
from lib_frame_qc import score_image_file, aggregate as qc_aggregate
PIPELINE_BASE = Path(os.environ.get("COSMA_PIPELINE_BASE", "/home/cosma/cosma-pipeline"))
QC_SAMPLE_RATE = int(os.environ.get("COSMA_QC_SAMPLE_RATE", "5"))
QC_BOTTOM_OK_PCT = float(os.environ.get("COSMA_QC_BOTTOM_OK_PCT", "50"))
NEED_STREAK = 10 # consecutive underwater frames required to lock start/end
# -----------------------------------------------------------------------------
# Trim logic (ported verbatim from dispatcher._AUTO_TRIM_SCRIPT)
# -----------------------------------------------------------------------------
def is_underwater(path: Path) -> bool | None:
img = cv2.imread(str(path), cv2.IMREAD_REDUCED_COLOR_4)
if img is None:
return None
b, g, r = [float(c) for c in cv2.mean(img)[:3]]
# Red is absorbed by water → R < G AND R < B on underwater shots.
return r < g - 5 and r < b - 5
def trim_segment(frames_dir: Path, dry_run: bool = False) -> tuple[int, int, int]:
"""Delete leading and trailing out-of-water frames.
Returns (head_removed, tail_removed, remaining).
"""
paths = sorted(frames_dir.glob("frame_*.jpg"))
if not paths:
return (0, 0, 0)
# Scan from start
start = 0
streak = 0
for i, p in enumerate(paths):
uw = is_underwater(p)
if uw is None:
continue
if uw:
streak += 1
if streak >= NEED_STREAK:
start = i - NEED_STREAK + 1
break
else:
streak = 0
# Scan from end
end = len(paths)
streak = 0
for j in range(len(paths) - 1, -1, -1):
uw = is_underwater(paths[j])
if uw is None:
continue
if uw:
streak += 1
if streak >= NEED_STREAK:
end = j + NEED_STREAK # exclusive
break
else:
streak = 0
if end <= start:
# Sanity: never delete everything.
start = 0
end = len(paths)
removed_head = start
removed_tail = len(paths) - end
if not dry_run:
for p in paths[:start]:
try:
p.unlink()
except OSError:
pass
for p in paths[end:]:
try:
p.unlink()
except OSError:
pass
return (removed_head, removed_tail, end - start)
# -----------------------------------------------------------------------------
# Safety: is this segment currently being touched?
# -----------------------------------------------------------------------------
def has_ffmpeg_running_on(frames_dir: Path) -> bool:
"""Check if any ffmpeg process is writing into frames_dir."""
try:
r = subprocess.run(
["pgrep", "-af", "ffmpeg"], capture_output=True, text=True, timeout=5
)
for line in r.stdout.splitlines():
if str(frames_dir) in line:
return True
except Exception:
pass
return False
def has_inference_running_on(frames_dir: Path) -> bool:
"""Check if any 05_inference.py process is running on frames_dir."""
try:
r = subprocess.run(
["pgrep", "-af", "05_inference"], capture_output=True, text=True, timeout=5
)
for line in r.stdout.splitlines():
if str(frames_dir) in line:
return True
except Exception:
pass
return False
def has_pending_inference_job(conn, mission_id: int, auv_id: str, segment: str) -> bool:
"""Check state.db for queued/running 05_inference job on this segment."""
row = conn.execute(
"SELECT status FROM jobs WHERE mission_id=? AND auv_id=? "
"AND segment_label=? AND stage='05_inference'",
(mission_id, auv_id, segment),
).fetchone()
if row is None:
return False
return row["status"] in ("queued", "running")
def frame_count_is_stable(frames_dir: Path, wait_s: float = 5.0) -> bool:
"""Return True if the frame count doesn't change over wait_s."""
n1 = sum(1 for _ in frames_dir.glob("frame_*.jpg"))
time.sleep(wait_s)
n2 = sum(1 for _ in frames_dir.glob("frame_*.jpg"))
return n1 == n2
# -----------------------------------------------------------------------------
# QC re-scoring (mirrors stage 04 qc_segment)
# -----------------------------------------------------------------------------
def qc_segment(frames_dir: Path, sample_rate: int = QC_SAMPLE_RATE) -> dict | None:
frames = sorted(frames_dir.glob("frame_*.jpg"))
if not frames:
return None
sampled = frames[::max(1, sample_rate)]
per_frame = []
for f in sampled:
s = score_image_file(f)
if s is not None:
per_frame.append(s)
if not per_frame:
return None
agg = qc_aggregate(per_frame)
qc_payload = {
"frames_in_dir": len(frames),
"frames_sampled": len(per_frame),
"sample_rate": sample_rate,
**agg,
"per_frame": per_frame,
}
try:
(frames_dir / "qc.json").write_text(json.dumps(qc_payload, indent=2))
except Exception as e:
print(f" [04b] qc.json write failed: {e}")
return agg
# -----------------------------------------------------------------------------
# Main per-segment driver
# -----------------------------------------------------------------------------
def process_segment(mission_name: str, auv_id: str, segment: str,
frames_dir: Path, dry_run: bool, conn) -> dict:
result = {
"auv_id": auv_id,
"segment": segment,
"frames_dir": str(frames_dir),
"skipped": False,
"head_removed": 0,
"tail_removed": 0,
"remaining": 0,
"before_total": 0,
"before_bottom_pct": None,
"after_bottom_pct": None,
"status_before": None,
"status_after": None,
}
if not frames_dir.is_dir():
result["skipped"] = True
result["reason"] = "no_frames_dir"
return result
# Safety checks
if has_ffmpeg_running_on(frames_dir):
result["skipped"] = True
result["reason"] = "ffmpeg_running"
print(f" [04b] SKIP {auv_id}/{segment}: ffmpeg still extracting")
return result
if has_inference_running_on(frames_dir):
result["skipped"] = True
result["reason"] = "inference_running_proc"
print(f" [04b] SKIP {auv_id}/{segment}: 05_inference process running")
return result
# Look up mission_id + current 04 job
mission_row = conn.execute(
"SELECT id FROM missions WHERE name=?", (mission_name,)
).fetchone()
if not mission_row:
result["skipped"] = True
result["reason"] = "mission_not_in_db"
return result
mission_id = mission_row["id"]
if has_pending_inference_job(conn, mission_id, auv_id, segment):
result["skipped"] = True
result["reason"] = "inference_job_pending"
print(f" [04b] SKIP {auv_id}/{segment}: 05_inference queued/running in DB")
return result
if not frame_count_is_stable(frames_dir, wait_s=5.0):
result["skipped"] = True
result["reason"] = "frame_count_unstable"
print(f" [04b] SKIP {auv_id}/{segment}: frame count not stable")
return result
# Snapshot before
before_paths = sorted(frames_dir.glob("frame_*.jpg"))
result["before_total"] = len(before_paths)
job04_row = conn.execute(
"SELECT id, status FROM jobs WHERE mission_id=? AND auv_id=? "
"AND segment_label=? AND stage='04_frame_extract'",
(mission_id, auv_id, segment),
).fetchone()
if job04_row is None:
result["skipped"] = True
result["reason"] = "no_04_job_in_db"
print(f" [04b] SKIP {auv_id}/{segment}: no 04 job row")
return result
result["status_before"] = job04_row["status"]
# Read current QC if available
qc_path = frames_dir / "qc.json"
if qc_path.exists():
try:
result["before_bottom_pct"] = json.loads(qc_path.read_text()).get("bottom_visible_pct")
except Exception:
pass
# Trim
head, tail, remaining = trim_segment(frames_dir, dry_run=dry_run)
result["head_removed"] = head
result["tail_removed"] = tail
result["remaining"] = remaining
# Re-QC if not dry-run and something was trimmed (or always to keep metrics fresh)
after_agg = None
if not dry_run and (head > 0 or tail > 0):
after_agg = qc_segment(frames_dir)
elif dry_run:
# In dry-run, don't touch qc.json; compute aggregate from remaining slice in-memory
remaining_paths = sorted(frames_dir.glob("frame_*.jpg"))[head: len(before_paths) - tail]
sampled = remaining_paths[::max(1, QC_SAMPLE_RATE)]
per_frame = [s for s in (score_image_file(f) for f in sampled) if s is not None]
if per_frame:
after_agg = qc_aggregate(per_frame)
if after_agg is not None:
result["after_bottom_pct"] = after_agg.get("bottom_visible_pct")
if dry_run:
print(
f" [04b] DRY {auv_id}/{segment}: head={head} tail={tail} "
f"remaining={remaining} (before={len(before_paths)}, "
f"bottom_pct {result['before_bottom_pct']}{result['after_bottom_pct']})"
)
return result
# Update DB: job row + metrics
job_id = job04_row["id"]
bottom_pct = after_agg.get("bottom_visible_pct") if after_agg else None
if bottom_pct is not None and bottom_pct >= QC_BOTTOM_OK_PCT:
new_status = "done"
err_msg = None
elif bottom_pct is not None:
new_status = "degraded"
err_msg = f"bottom_visible_pct={bottom_pct}% <{QC_BOTTOM_OK_PCT}% (after trim)"
else:
new_status = job04_row["status"]
err_msg = None
upsert_job(
conn, mission_id, auv_id, segment, "04_frame_extract",
status=new_status,
output_path=str(frames_dir),
error_msg=err_msg,
)
record_metric(conn, job_id, "trimmed_head", value=float(head))
record_metric(conn, job_id, "trimmed_tail", value=float(tail))
record_metric(conn, job_id, "frames_after_trim", value=float(remaining))
if after_agg:
for k in (
"frames_total", "frames_bottom_visible", "frames_out_of_water",
"frames_turbid", "frames_water_no_bottom",
):
if k in after_agg:
record_metric(conn, job_id, k, value=float(after_agg[k]))
if bottom_pct is not None:
record_metric(
conn, job_id, "bottom_visible_pct",
value=float(bottom_pct),
pass_fail="pass" if bottom_pct >= QC_BOTTOM_OK_PCT else "degraded",
)
result["status_after"] = new_status
print(
f" [04b] {auv_id}/{segment}: trimmed head={head} tail={tail} "
f"remaining={remaining}, bottom_pct={bottom_pct}% ({result['status_before']}{new_status})"
)
return result
# -----------------------------------------------------------------------------
# Discovery + CLI
# -----------------------------------------------------------------------------
def find_segments(mission_name: str, auv_filter: str | None,
segment_filter: str | None) -> list[tuple[str, str, Path]]:
base = PIPELINE_BASE / "data" / mission_name / "frames"
out: list[tuple[str, str, Path]] = []
if not base.is_dir():
return out
for auv_dir in sorted(base.iterdir()):
if not auv_dir.is_dir():
continue
if auv_filter and auv_dir.name != auv_filter:
continue
for seg_dir in sorted(auv_dir.iterdir()):
if not seg_dir.is_dir():
continue
if segment_filter and seg_dir.name != segment_filter:
continue
out.append((auv_dir.name, seg_dir.name, seg_dir))
return out
def main():
ap = argparse.ArgumentParser(description="Stage 04b — Trim hors-eau head/tail frames")
ap.add_argument("--mission", default="20260505-Lepradet")
ap.add_argument("--auv")
ap.add_argument("--segment")
ap.add_argument("--dry-run", action="store_true")
args = ap.parse_args()
init_db()
segments = find_segments(args.mission, args.auv, args.segment)
if not segments:
print(f"[04b] No segments found under {args.mission}")
sys.exit(1)
print(f"[04b] Mission={args.mission} segments={len(segments)} dry_run={args.dry_run}")
results: list[dict] = []
with get_conn() as conn:
for auv_id, segment, frames_dir in segments:
try:
r = process_segment(args.mission, auv_id, segment, frames_dir,
args.dry_run, conn)
except Exception as e:
r = {"auv_id": auv_id, "segment": segment, "error": str(e),
"skipped": True}
print(f" [04b] ERR {auv_id}/{segment}: {e}")
results.append(r)
# Summary
print("\n=== Stage 04b summary ===")
upgraded = [r for r in results
if r.get("status_before") == "degraded" and r.get("status_after") == "done"]
still_degraded = [r for r in results
if r.get("status_after") == "degraded"]
skipped = [r for r in results if r.get("skipped")]
print(f"Upgraded degraded→done : {len(upgraded)}")
for r in upgraded:
print(f" + {r['auv_id']}/{r['segment']} "
f"({r['before_bottom_pct']}%→{r['after_bottom_pct']}%, "
f"trim head={r['head_removed']} tail={r['tail_removed']})")
print(f"Still degraded : {len(still_degraded)}")
print(f"Skipped : {len(skipped)}")
for r in skipped:
print(f" - {r['auv_id']}/{r['segment']}: {r.get('reason', 'unknown')}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,376 @@
#!/usr/bin/env python3
"""Stage 05 — Run lingbot-map inference on extracted frames.
Args:
--frames-dir <path> Directory with frame_*.jpg (or parent with AUV subdirs)
--worker <auto|.84|.87> GPU worker selection
--mission <name> Mission name for output paths
Workers:
.84: /root/ai-video/lingbot-map/.venv/bin/python demo.py ...
.87: /home/floppyrj45/ai-video/lingbot-map/.venv/bin/python demo.py ...
Auto: pick by lowest GPU memory usage (nvidia-smi via SSH).
Flow:
1. Kill any stale demo.py on worker before starting
2. rsync frames .83 → worker /root/cosma-frames-tmp/
3. SSH launch demo.py in background; poll for PLY file; kill viser server once PLY done
4. Retrieve PLY + NPZ → .83 ~/cosma-pipeline/data/<mission>/ply/<AUV>/<segment>.{ply,npz}
5. Cleanup worker temp dir
6. Log to SQLite: duration, GPU peak mem, nb points in PLY
Usage:
python3 05_inference.py --frames-dir ~/cosma-pipeline/data/20260505-Lepradet/frames/AUV210/GX019837 --worker auto --mission 20260505-Lepradet
"""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
import time
from pathlib import Path
import yaml
sys.path.insert(0, str(Path(__file__).parent.parent))
from orchestrator.db import init_db, get_conn, upsert_job, record_metric, now_iso
PIPELINE_BASE = Path(os.environ.get("COSMA_PIPELINE_BASE", "/home/cosma/cosma-pipeline"))
def _load_inference_cfg() -> dict:
"""Load inference params from thresholds.yaml, with sane defaults."""
cfg_path = Path(__file__).parent.parent / "config" / "thresholds.yaml"
try:
data = yaml.safe_load(cfg_path.read_text())
return data.get("inference", {})
except Exception:
return {}
_INF_CFG = _load_inference_cfg()
WORKERS = {
".84": {
"host": "192.168.0.84",
"user": "root",
"ai_dir": "/root/ai-video/lingbot-map",
"venv": "/root/ai-video/lingbot-map/.venv/bin/python",
"tmp_dir": "/root/cosma-frames-tmp",
},
".87": {
"host": "192.168.0.87",
"user": "floppyrj45",
"ai_dir": "/home/floppyrj45/ai-video/lingbot-map",
"venv": "/home/floppyrj45/ai-video/lingbot-map/.venv/bin/python",
"tmp_dir": "/home/floppyrj45/cosma-frames-tmp",
},
}
def get_gpu_mem_used(worker_key: str) -> int:
"""Return GPU memory used in MB via SSH nvidia-smi. Returns 99999 on error."""
w = WORKERS[worker_key]
cmd = [
"ssh", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=5",
f"{w['user']}@{w['host']}",
"nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits 2>/dev/null | head -1"
]
try:
r = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
return int(r.stdout.strip())
except Exception:
return 99999
def kill_stale_demo_py(worker_key: str) -> None:
"""Kill any lingering demo.py processes on worker before starting new inference."""
w = WORKERS[worker_key]
ssh_target = f"{w['user']}@{w['host']}"
try:
subprocess.run(
["ssh", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=10",
ssh_target, "pkill -9 -f demo.py 2>/dev/null; sleep 1; echo stale_killed"],
capture_output=True, text=True, timeout=15,
)
print(f" [05] Stale demo.py killed on {worker_key}")
except Exception as e:
print(f" [05] Warning: kill_stale failed on {worker_key}: {e}")
def pick_worker() -> str:
"""Auto-select worker with lowest GPU memory usage."""
best = None
best_mem = 99999
for key in WORKERS:
mem = get_gpu_mem_used(key)
print(f" [05] Worker {key}: GPU mem={mem}MB")
if mem < best_mem:
best_mem = mem
best = key
if best is None:
raise RuntimeError("No GPU worker available")
print(f" [05] Selected worker {best}")
return best
def count_ply_points(ply_path: Path) -> int:
"""Count vertex count in PLY file header."""
try:
with open(ply_path, "rb") as f:
for _ in range(30):
line = f.readline().decode("ascii", errors="ignore").strip()
if line.startswith("element vertex"):
return int(line.split()[-1])
except Exception:
pass
return 0
def run_inference(frames_dir: Path, worker_key: str, mission_name: str,
auv_id: str, segment: str) -> dict:
"""Run lingbot-map on one segment. Returns metrics."""
w = WORKERS[worker_key]
host = w["host"]
user = w["user"]
ssh_target = f"{user}@{host}"
worker_frames = f"{w['tmp_dir']}/{mission_name}/{auv_id}/{segment}"
ply_remote = f"{w['tmp_dir']}/{mission_name}/{auv_id}/{segment}.ply"
npz_remote = f"{w['tmp_dir']}/{mission_name}/{auv_id}/{segment}.npz"
out_dir = PIPELINE_BASE / "data" / mission_name / "ply" / auv_id
out_dir.mkdir(parents=True, exist_ok=True)
out_ply = out_dir / f"{segment}.ply"
out_npz = out_dir / f"{segment}.npz"
if out_ply.exists() and out_ply.stat().st_size > 1000:
n_pts = count_ply_points(out_ply)
print(f" [05] {auv_id}/{segment}: cached PLY ({n_pts} pts)")
return {"cached": True, "ply": str(out_ply), "n_points": n_pts}
metrics = {
"auv_id": auv_id,
"segment": segment,
"worker": worker_key,
"status": "ok",
}
# Step 0: kill any stale demo.py on worker
kill_stale_demo_py(worker_key)
# Step 1: create remote temp dir + rsync frames
print(f" [05] rsync {frames_dir}{ssh_target}:{worker_frames}...")
subprocess.run(
["ssh", "-o", "StrictHostKeyChecking=no", ssh_target,
f"mkdir -p {worker_frames}"],
check=True, timeout=15,
)
r = subprocess.run(
["rsync", "-az", "--delete",
str(frames_dir) + "/",
f"{ssh_target}:{worker_frames}/"],
capture_output=True, text=True, timeout=600,
)
if r.returncode != 0:
metrics["status"] = "error"
metrics["error"] = f"rsync failed: {r.stderr[-200:]}"
return metrics
print(f" [05] rsync done")
# Step 2: build demo.py command -- params from thresholds.yaml[inference]
checkpoint = f"{w['ai_dir']}/checkpoints/lingbot-map/lingbot-map.pt"
inf_mode = _INF_CFG.get("mode", "streaming")
conf_thr = _INF_CFG.get("ply_conf_threshold", 1.5)
kf_interval = _INF_CFG.get("keyframe_interval", 1)
max_frames = _INF_CFG.get("max_frame_num", 1024)
use_offload = _INF_CFG.get("offload_to_cpu", False)
offload_flag = "--offload_to_cpu" if use_offload else "--no-offload_to_cpu"
if inf_mode == "windowed":
window_size = _INF_CFG.get("window_size", 64)
overlap_size = _INF_CFG.get("overlap_size", 16)
mode_flags = (
f"--mode windowed "
f"--window_size {window_size} "
f"--overlap_size {overlap_size} "
)
else: # streaming (default, validated GX049839_v2 146M pts)
mode_flags = (
f"--mode streaming "
f"--keyframe_interval {kf_interval} "
f"--max_frame_num {max_frames} "
)
inf_timeout = int(_INF_CFG.get("inference_timeout_s", 10800))
# Remote script: launch demo.py in background, poll for PLY, kill viser when done
# This avoids the SSH blocking on the viser server that starts after inference
remote_script = f"""#!/bin/bash
set -e
PLY={ply_remote}
LOG=/tmp/cosma_demo_{segment}.log
# Launch demo.py in background
nohup {w['venv']} {w['ai_dir']}/demo.py \\
--model_path {checkpoint} \\
--image_folder {worker_frames} \\
{mode_flags}--ply_conf_threshold {conf_thr} \\
--save_ply \\
--save_poses {npz_remote} \\
--use_sdpa {offload_flag} \\
> 2>&1 &
DEMO_PID=
echo "demo.py PID=" >&2
# Poll for PLY file (check every 30s)
WAITED=0
while [ -lt {inf_timeout} ]; do
if [ -f "" ] && [ $(wc -c < "") -gt 100 ]; then
sleep 10 # let write finish
echo "PLY_DONE size=$(wc -c < )" >&2
kill 2>/dev/null || true
exit 0
fi
# Check if process died with error
if ! kill -0 2>/dev/null; then
echo "Process died early" >&2
exit 1
fi
sleep 30
WAITED=30
done
echo "TIMEOUT after {inf_timeout}s" >&2
kill -9 2>/dev/null || true
exit 2
"""
print(f" [05] Launching inference on {host} (background+poll, timeout={inf_timeout}s)...")
t0 = time.time()
r = subprocess.run(
["ssh", "-o", "StrictHostKeyChecking=no", ssh_target,
"bash -s"],
input=remote_script,
capture_output=True, text=True, timeout=inf_timeout + 60,
)
elapsed = time.time() - t0
metrics["inference_s"] = round(elapsed, 1)
if r.returncode != 0:
metrics["status"] = "error"
metrics["error"] = (r.stdout + r.stderr)[-500:]
print(f" [05] inference error: {metrics['error'][-200:]}")
return metrics
print(f" [05] Inference done in {elapsed:.1f}s (returncode={r.returncode})")
metrics["gpu_peak_mb"] = get_gpu_mem_used(worker_key)
# Step 4: rsync PLY + NPZ back
print(f" [05] Retrieving PLY + NPZ...")
for remote, local in [(ply_remote, out_ply), (npz_remote, out_npz)]:
r2 = subprocess.run(
["rsync", "-az", f"{ssh_target}:{remote}", str(local)],
capture_output=True, text=True, timeout=120,
)
if r2.returncode != 0:
print(f" [05] Warning: rsync back failed for {remote}: {r2.stderr[-100:]}")
# Step 5: cleanup worker
subprocess.run(
["ssh", "-o", "StrictHostKeyChecking=no", ssh_target,
f"rm -rf {worker_frames} {ply_remote} {npz_remote}"],
timeout=30,
)
# Count PLY points
n_pts = count_ply_points(out_ply) if out_ply.exists() else 0
metrics["n_points"] = n_pts
metrics["ply"] = str(out_ply)
print(f" [05] PLY: {n_pts} points → {out_ply}")
return metrics
def process_frames_dir(frames_dir: Path, worker_key: str, mission_name: str) -> list[dict]:
"""Process a directory of frames (single segment or AUV tree)."""
direct_frames = list(frames_dir.glob("frame_*.jpg"))
if direct_frames:
parts = frames_dir.parts
auv_id = frames_dir.parent.name if len(parts) >= 2 else "UNKNOWN"
segment = frames_dir.name
return [run_inference(frames_dir, worker_key, mission_name, auv_id, segment)]
all_metrics = []
for auv_dir in sorted(frames_dir.iterdir()):
if not auv_dir.is_dir():
continue
auv_id = auv_dir.name
for seg_dir in sorted(auv_dir.iterdir()):
if not seg_dir.is_dir():
continue
frames = list(seg_dir.glob("frame_*.jpg"))
if not frames:
continue
print(f"\n[05] === {auv_id}/{seg_dir.name}: {len(frames)} frames ===")
# Guard: min frames required for model (RoPE/attention)
min_frames = int(_INF_CFG.get("min_frames_for_inference", 32))
if len(frames) < min_frames:
print(f" [05] SKIP {auv_id}/{seg_dir.name}: {len(frames)} frames < {min_frames} min")
init_db()
with get_conn() as conn_mf:
mr = conn_mf.execute("SELECT id FROM missions WHERE name=?", (mission_name,)).fetchone()
if mr:
upsert_job(conn_mf, mr["id"], auv_id, seg_dir.name, "05_inference",
status="skipped",
error_msg=f"frames_too_few={len(frames)}<{min_frames}")
continue
m = run_inference(seg_dir, worker_key, mission_name, auv_id, seg_dir.name)
all_metrics.append(m)
init_db()
with get_conn() as conn:
mission_row = conn.execute(
"SELECT id FROM missions WHERE name=?", (mission_name,)
).fetchone()
if mission_row and not m.get("cached"):
job_id = upsert_job(
conn, mission_row["id"], auv_id, seg_dir.name, "05_inference",
status="done" if m.get("status") == "ok" else m.get("status", "error"),
output_path=m.get("ply", ""),
)
record_metric(conn, job_id, "ply_points", value=m.get("n_points", 0),
pass_fail="pass" if m.get("n_points", 0) > 100 else "fail")
if "inference_s" in m:
record_metric(conn, job_id, "inference_s", value=m["inference_s"])
if "gpu_peak_mb" in m:
record_metric(conn, job_id, "gpu_peak_mb", value=m["gpu_peak_mb"])
return all_metrics
def main():
ap = argparse.ArgumentParser(description="Stage 05 — lingbot-map inference")
ap.add_argument("--frames-dir", type=Path, required=True)
ap.add_argument("--worker", type=str, default="auto", choices=["auto", ".84", ".87"])
ap.add_argument("--mission", type=str, required=True)
args = ap.parse_args()
worker = args.worker
if worker == "auto":
worker = pick_worker()
metrics = process_frames_dir(args.frames_dir, worker, args.mission)
print("\n=== Stage 05 summary ===")
total_pts = sum(m.get("n_points", 0) for m in metrics)
ok = sum(1 for m in metrics if m.get("status") == "ok" or m.get("cached"))
print(f" Segments OK: {ok}/{len(metrics)}, total PLY points: {total_pts}")
for m in metrics:
print(f" {m.get('auv_id','?')}/{m.get('segment','?')}: "
f"{m.get('n_points',0)} pts "
f"[{m.get('status','cached' if m.get('cached') else '?')}]")
if __name__ == "__main__":
main()

View File

View File

@@ -0,0 +1,83 @@
"""Frame quality scoring for underwater footage.
For each frame, compute:
- laplacian_var: focus/sharpness (cv2.Laplacian variance)
- contrast: stddev of grayscale
- blue_dominance: mean(B - R), positive = water dominant
- mean_r/g/b: per-channel means
Classification (priority order):
- mean_r > mean_g + 5 AND mean_r > mean_b + 5 → 'out_of_water'
- laplacian_var < 50 AND contrast < 25 → 'turbid_water'
- laplacian_var >= 80 AND contrast >= 35
AND blue_dominance > -10 → 'bottom_visible'
- else → 'water_no_bottom'
"""
from __future__ import annotations
from collections import Counter
from typing import Iterable
import cv2
import numpy as np
def score_frame(frame_bgr: np.ndarray) -> dict:
"""Return per-frame QC metrics + class label."""
gray = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2GRAY)
lap_var = float(cv2.Laplacian(gray, cv2.CV_64F).var())
contrast = float(gray.std())
b, g, r = cv2.split(frame_bgr)
mean_r = float(r.mean())
mean_g = float(g.mean())
mean_b = float(b.mean())
blue_dom = float(mean_b - mean_r)
if mean_r > mean_g + 5 and mean_r > mean_b + 5:
klass = "out_of_water"
elif lap_var < 50 and contrast < 25:
klass = "turbid_water"
elif lap_var >= 80 and contrast >= 35 and blue_dom > -10:
klass = "bottom_visible"
else:
klass = "water_no_bottom"
return {
"laplacian_var": round(lap_var, 2),
"contrast": round(contrast, 2),
"blue_dominance": round(blue_dom, 2),
"mean_r": round(mean_r, 1),
"mean_g": round(mean_g, 1),
"mean_b": round(mean_b, 1),
"class": klass,
"score_ok": klass == "bottom_visible",
}
def score_image_file(path) -> dict | None:
"""Load image with OpenCV and score it. Returns None on failure."""
img = cv2.imread(str(path))
if img is None:
return None
res = score_frame(img)
res["file"] = str(path)
return res
def aggregate(scores: Iterable[dict]) -> dict:
"""Aggregate a sequence of score_frame() dicts."""
scores = list(scores)
total = len(scores)
counts = Counter(s["class"] for s in scores)
bottom = counts.get("bottom_visible", 0)
return {
"frames_total": total,
"frames_bottom_visible": bottom,
"frames_out_of_water": counts.get("out_of_water", 0),
"frames_turbid": counts.get("turbid_water", 0),
"frames_water_no_bottom": counts.get("water_no_bottom", 0),
"bottom_visible_pct": round(100.0 * bottom / total, 1) if total else 0.0,
}
CLASS_ORDER = ("bottom_visible", "water_no_bottom", "turbid_water", "out_of_water")

View File

@@ -0,0 +1,14 @@
# Veille — 2026-05-11 22:33 UTC — iter-1
## ArXiv (signaux forts)
- **[2605.04672]** AI-Aided Advancements in AUV Navigation — fusion caméra+DVL+IMU IA, pertinent nav AUV
- **BALTIC benchmark** — cross-domain 3D recon air/eau illumination variable
- **3D Gaussian Splatting underwater** — spatiotemporal degradation-aware GS scènes turbides
## GitHub
- **LingBot-Map** (maj 3j) — streaming 20FPS 518×378 10k+ frames drift correction attention paginée — fort signal
- **DUSt3R** actif, suivi normal
- MonST3R / VGGT / MoGe : pas de maj 7j
## Action possible
Tester 3DGS underwater sur frames AUV210 turbides (0% bottom visible) comme alternative à lingbot reconstruction

View File

@@ -0,0 +1,23 @@
# Veille COSMA reconstruction — iter-2 — 2026-05-12 04:30 UTC
## arxiv underwater 3D (7 derniers jours)
- UW-3DGS: Underwater 3D Reconstruction, Physics-Aware Gaussian Splatting (arxiv 2508.06169)
- Visual enhancement + 3D representation underwater: review (arxiv 2505.01869)
## arxiv AUV SLAM / point cloud
- VISO: Robust Underwater Visual-Inertial-Sonar SLAM (arxiv 2601.01144) — VIS+sonar, fort intérêt pour pipeline USBL
- RUSSO: Underwater SLAM stéréo+sonar+IMU (arxiv 2503.01434)
- VIMS: Visual-Inertial-Magnetic-Sonar SLAM (arxiv 2506.15126)
## Repos GitHub actifs
- naver/dust3r (7k★): actif, base pipeline lingbot-map
- Junyi42/monst3r (ICLR 2025): géométrie vidéo dynamique
- facebookresearch/vggt (CVPR 2025 Best Paper): reconstruction per-frame
- CUT3R: Continuous 3D Perception, mise à jour mars 2026
## HuggingFace
- Video-Depth-Anything-Small: depth video temps-réel
- StereoAdapter: adaptation profondeur stéréo sous-marine
## Signal fort
VISO (arxiv 2601.01144): pipeline USBL+caméra+IMU pour AUV, pourrait remplacer pure-camera pose estimation dans stage 06_align.

View File

@@ -0,0 +1,26 @@
# Veille iter-4 — 2026-05-12 16:50 UTC
## Top signaux (8-9/10)
- **ReefMapGS** arxiv.org/abs/2604.11992 — SLAM+3DGS 700m AUV, COLMAP-free, directement applicable COSMA (9/10)
- **OceanSplat** (2026) — 3D Gaussian Splatting milieu turbide + trinocular consistency (9/10)
- **BIND-USBL** arxiv.org/abs/2604.11861 — fusion IMU+USBL hétérogène ASV-AUV, delayed fusion = pattern réutilisable stage 06_align (9/10)
- **LingBot-Map update** (27 avril) — keyframe_interval fix + long-video demo — update recommandé (8/10)
- **PAS3R** HuggingFace — Pose-Adaptive Streaming 3D, long video = streaming AUV (8/10)
- **AI-Aided AUV Navigation** arxiv.org/abs/2605.04672 — fusion INS+DVL+cam deep learning (8/10)
## Signaux modérés (7/10)
- Aquatic Neuromorphic Optical Flow arxiv.org/abs/2605.07653 — event cam AUV turbide
- WaterSplat-SLAM RAL 2026 — SLAM monoculaire sous-marin photoréaliste
## Repos actifs
- lingbot-map (keyframe fix avril), awesome-dust3r (ecosystem DUSt3R/VGGT/CUT3R)
- Matisse Ifremer — datasets flotte française
## Recommandations
1. **BIND-USBL** : lire pour stage 06_align (pattern fusion USBL+IMU déjà dispo)
2. **LingBot-Map update** : Already up to date. sur .84/.87 avant prochaine iter
3. **ReefMapGS** : évaluer comme alternative stage 06_align si PR #9/#12 mergés

View File

@@ -0,0 +1,26 @@
# Veille Iter-5 — 2026-05-12 22:46 UTC
## Arxiv / Papers
| # | Titre | Signal | Score |
|---|-------|--------|-------|
| 1 | ReefMapGS | SLAM multimodal + Gaussian Splatting pour grandes scènes sous-marines avec fermeture de boucle | 9/10 |
| 2 | Sonar-MASt3R | Fusion optico-acoustique temps réel pour environnements turbides — intéressant pour milieu turbide AUV | 8/10 |
| 3 | WaterSplat-SLAM | SLAM monoculaire photoréaliste underwater, moindre dépendance stéréo | 8/10 |
| 4 | Spatiotemporal Degradation-Aware 3DGS | Reconstruction scènes sous-marines avec dégradation temporelle (particules, courant) | 8/10 |
| 5 | BALTIC Benchmark | Benchmark 3D reconstruction air/underwater avec variations d'illumination, utile pour QC comparaison | 7/10 |
| 6 | Lost at Sea (Notre Dame) | AUV utilisant 3DGS pour navigation autonome et reconnaissance environnement | 7/10 |
## GitHub / HuggingFace
| Repo | Signal |
|------|--------|
| LingBot-Map | Commits récents (4 jours) — à tracker pour keyframe fixes |
| dust3r/mast3r | Actifs, pas de release majeure dernière semaine |
| Pixal3D (SIGGRAPH 2026) | 3D pixel-alignée, potentiellement utile pour poses denses |
## Recommandation prochaine iteration
- **ReefMapGS** : évaluer pour remplacement LingBot-Map sur grands segments (15m+)
- **Sonar-MASt3R** : pertinent si Kogger SBP intégré dans pipeline — stage 06 USBL+cam pourrait utiliser composante acoustique
- **BALTIC Benchmark** : utiliser pour QC comparatif sur segments AUV210 (turbide)

View File

@@ -0,0 +1,21 @@
# Veille iter-7 — 2026-05-13 10:43 UTC
## Papers / Signaux (6 total)
| # | Titre | Ref | Score | Pertinence COSMA |
|---|-------|-----|-------|-----------------|
| 1 | Aquatic Neuromorphic Optical Flow | arXiv 2605.07653 (5j) | 9/10 | Optique turbide robuste, temps-réel, léger → stage06_align |
| 2 | MAGS-SLAM: Multi-Agent 3DGS SLAM | arXiv 2605.10760 (2j) | 8/10 | SLAM 3DGS multi-robot, cohérence photométrique → futur multi-AUV |
| 3 | AI Platform AUV 3DGS (Notre-Dame) | engineering.nd.edu (5j) | 9/10 | 3DGS ellipsoïdes flous underwater, navigation AUV pré-chargée |
| 4 | MV-DUSt3R+ | GitHub facebookresearch (7j) | 8/10 | DUSt3R v2 rapide (2s), baseline comparaison stage05 |
| 5 | MonST3R | GitHub Junyi42 (ICLR 2025) | 7/10 | Géométrie robuste motion/occlusion → transition segments |
| 6 | LingBot-Map | GitHub robbyant (5j) | 9/10 | Màj streaming, vérifier diff vs version .84/.87 installée |
## Repos actifs (7j)
- **lingbot-map** (robbyant) : dernière màj 5j — comparer avec version installée .84/.87
- **dust3r / monst3r** : mises à jour README et poids — rien d'urgent
## Recommandations prochaines
1. Évaluer Aquatic Neuromorphic Optical Flow pour stage06_align (turbide)
2. Benchmarker 3DGS (MAGS-SLAM ou Notre-Dame) sur 1 segment AUV210
3. Mettre à jour lingbot-map .84/.87 si diff significatif

72
scripts/viser_auv.py Normal file
View File

@@ -0,0 +1,72 @@
#!/usr/bin/env python3
"""Open viser viewer with all PLYs from one AUV.
Usage:
viser_auv.py --ply-dir /path/to/auv/ply --port 9210
"""
from __future__ import annotations
import argparse
import sys
import time
from pathlib import Path
import numpy as np
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--ply-dir", required=True)
ap.add_argument("--port", type=int, default=9210)
ap.add_argument("--point-size", type=float, default=0.01)
ap.add_argument("--max-points-per-ply", type=int, default=1_500_000)
args = ap.parse_args()
try:
import open3d as o3d
import viser
except ImportError as e:
sys.exit(f"missing dep: {e}")
ply_dir = Path(args.ply_dir)
plys = sorted(ply_dir.glob("**/*.ply"))
print(f"Found {len(plys)} PLY files in {ply_dir}", flush=True)
if not plys:
sys.exit("no PLY found")
server = viser.ViserServer(host="0.0.0.0", port=args.port)
palette = [
(1.0, 0.30, 0.30), (0.30, 1.0, 0.30), (0.30, 0.55, 1.0),
(1.0, 0.85, 0.20), (1.0, 0.30, 1.0), (0.30, 1.0, 1.0),
(1.0, 0.55, 0.20), (0.55, 0.30, 1.0),
]
for i, p in enumerate(plys):
pcd = o3d.io.read_point_cloud(str(p))
pts = np.asarray(pcd.points, dtype=np.float32)
if len(pts) == 0:
print(f" ! {p.name}: empty", flush=True)
continue
if pcd.has_colors():
cols = np.asarray(pcd.colors, dtype=np.float32)
else:
cols = np.tile(palette[i % len(palette)], (len(pts), 1)).astype(np.float32)
if len(pts) > args.max_points_per_ply:
idx = np.random.choice(len(pts), args.max_points_per_ply, replace=False)
pts = pts[idx]
cols = cols[idx]
# viser wants uint8 colors
cols_u8 = (cols * 255).clip(0, 255).astype(np.uint8)
name = f"/{p.parent.name}_{p.stem}"
server.scene.add_point_cloud(
name=name, points=pts, colors=cols_u8, point_size=args.point_size
)
print(f" + {p.name}: {len(pts):,} pts", flush=True)
print(f"Viser ready on port {args.port}", flush=True)
while True:
time.sleep(60)
if __name__ == "__main__":
main()