6a703c314d
Incomplete: step 1 (fixture markers + track_truth.json), step 6 (frontend wiring), step 8 (tests) not done; suite not yet run. Salvaged from interrupted prior session. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
696 lines
24 KiB
Python
696 lines
24 KiB
Python
"""SQLAlchemy engine/session and schema (spec §2) plus small CRUD helpers.
|
|
|
|
FROZEN CONTRACT after foundation: the schema (table/column names and types) and the
|
|
helper signatures are what every lane depends on. Lanes write through these helpers only.
|
|
|
|
Schema (spec §2)::
|
|
|
|
videos(id, filename, duration_s, fps, width, height, offset_ms, drift_ppm,
|
|
sync_confidence, created_at)
|
|
camera_poses(id, video_id FK, frame_idx, t_video_s, qw,qx,qy,qz, tx,ty,tz,
|
|
fx,fy,cx,cy, registered) # COLMAP world->camera; registered=False => interpolated
|
|
anchors(id, label, x,y,z, color)
|
|
events(id, t_global_s, duration_s, event_type, confidence, description, source)
|
|
annotations(id, video_id FK, t_video_s, x0,y0,x1,y1, resolved_anchor_id FK NULL)
|
|
paths(id, name, created_at, path_json) # phase 5 (M16): saved camPath JSON, as text
|
|
tracks(id, marker_key, label, color, created_at) # phase 6 (M18/M20): friend tracks
|
|
track_points(id, track_id FK, t_global_s, x,y,z, quality, views) # ordered by t_global_s
|
|
|
|
Engine management: :func:`init_engine` (re)binds the module to a SQLite file. It defaults
|
|
to ``config.DB_PATH`` but tests point it at a temp file. Helpers open and commit their own
|
|
short-lived sessions and return detached ORM instances (``expire_on_commit=False``), so the
|
|
returned objects are safe to read after the session closes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Iterable, Iterator, Sequence
|
|
|
|
from sqlalchemy import (
|
|
Boolean,
|
|
Float,
|
|
ForeignKey,
|
|
Integer,
|
|
String,
|
|
create_engine,
|
|
delete,
|
|
select,
|
|
update,
|
|
)
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker
|
|
|
|
from festival4d import config
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ORM models
|
|
# ---------------------------------------------------------------------------
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
class Video(Base):
|
|
__tablename__ = "videos"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
filename: Mapped[str] = mapped_column(String, nullable=False, unique=True)
|
|
duration_s: Mapped[float] = mapped_column(Float, nullable=False)
|
|
fps: Mapped[float] = mapped_column(Float, nullable=False)
|
|
width: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
height: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
offset_ms: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
drift_ppm: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
sync_confidence: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
created_at: Mapped[str] = mapped_column(String, nullable=False)
|
|
|
|
|
|
class CameraPose(Base):
|
|
__tablename__ = "camera_poses"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
video_id: Mapped[int] = mapped_column(ForeignKey("videos.id"), nullable=False, index=True)
|
|
frame_idx: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
t_video_s: Mapped[float] = mapped_column(Float, nullable=False)
|
|
# COLMAP world->camera quaternion [w,x,y,z] + translation
|
|
qw: Mapped[float] = mapped_column(Float, nullable=False)
|
|
qx: Mapped[float] = mapped_column(Float, nullable=False)
|
|
qy: Mapped[float] = mapped_column(Float, nullable=False)
|
|
qz: Mapped[float] = mapped_column(Float, nullable=False)
|
|
tx: Mapped[float] = mapped_column(Float, nullable=False)
|
|
ty: Mapped[float] = mapped_column(Float, nullable=False)
|
|
tz: Mapped[float] = mapped_column(Float, nullable=False)
|
|
# pinhole intrinsics after undistortion
|
|
fx: Mapped[float] = mapped_column(Float, nullable=False)
|
|
fy: Mapped[float] = mapped_column(Float, nullable=False)
|
|
cx: Mapped[float] = mapped_column(Float, nullable=False)
|
|
cy: Mapped[float] = mapped_column(Float, nullable=False)
|
|
registered: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
|
|
|
|
|
|
class Anchor(Base):
|
|
__tablename__ = "anchors"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
label: Mapped[str] = mapped_column(String, nullable=False)
|
|
x: Mapped[float] = mapped_column(Float, nullable=False)
|
|
y: Mapped[float] = mapped_column(Float, nullable=False)
|
|
z: Mapped[float] = mapped_column(Float, nullable=False)
|
|
color: Mapped[str | None] = mapped_column(String, nullable=True)
|
|
|
|
|
|
class Event(Base):
|
|
__tablename__ = "events"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
t_global_s: Mapped[float] = mapped_column(Float, nullable=False)
|
|
duration_s: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
event_type: Mapped[str] = mapped_column(String, nullable=False)
|
|
confidence: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
description: Mapped[str | None] = mapped_column(String, nullable=True)
|
|
source: Mapped[str] = mapped_column(String, nullable=False) # 'audio_auto' | 'ai' | 'user'
|
|
|
|
|
|
class Annotation(Base):
|
|
__tablename__ = "annotations"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
video_id: Mapped[int] = mapped_column(ForeignKey("videos.id"), nullable=False, index=True)
|
|
t_video_s: Mapped[float] = mapped_column(Float, nullable=False)
|
|
x0: Mapped[float] = mapped_column(Float, nullable=False)
|
|
y0: Mapped[float] = mapped_column(Float, nullable=False)
|
|
x1: Mapped[float] = mapped_column(Float, nullable=False)
|
|
y1: Mapped[float] = mapped_column(Float, nullable=False)
|
|
# Nullable link to the event this annotation locates (CR-2, M8). Two annotations sharing an
|
|
# event_id from different videos triangulate to a 3D anchor. Nullable => additive/back-compat.
|
|
event_id: Mapped[int | None] = mapped_column(ForeignKey("events.id"), nullable=True)
|
|
resolved_anchor_id: Mapped[int | None] = mapped_column(
|
|
ForeignKey("anchors.id"), nullable=True
|
|
)
|
|
|
|
|
|
class SavedPath(Base):
|
|
"""A saved keyframed camera path (phase 5 contract #3). ``path_json`` is the frozen
|
|
camPath JSON (``{"version": 1, "keyframes": [...]}``) stored verbatim as text —
|
|
the DB never interprets it beyond POST-time validation in the API layer."""
|
|
|
|
__tablename__ = "paths"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
name: Mapped[str] = mapped_column(String, nullable=False)
|
|
created_at: Mapped[str] = mapped_column(String, nullable=False)
|
|
path_json: Mapped[str] = mapped_column(String, nullable=False)
|
|
|
|
|
|
class Track(Base):
|
|
"""A friend track (phase 6 contract #2): a wearable marker's identity + metadata.
|
|
|
|
``marker_key`` is the detector-emitted identity — ``"hue:<opencv_hue>"`` for a color
|
|
marker or ``"code:<id>"`` for a blink badge (see ``tracker_detect``). The per-timestep
|
|
3D path lives in :class:`TrackPoint`. ``label`` / ``color`` are user-editable.
|
|
"""
|
|
|
|
__tablename__ = "tracks"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
marker_key: Mapped[str] = mapped_column(String, nullable=False)
|
|
label: Mapped[str | None] = mapped_column(String, nullable=True)
|
|
color: Mapped[str | None] = mapped_column(String, nullable=True)
|
|
created_at: Mapped[str] = mapped_column(String, nullable=False)
|
|
|
|
|
|
class TrackPoint(Base):
|
|
"""One solved position of a track at a master-timeline instant (phase 6 contract #2).
|
|
|
|
Coordinates are **Three.js scene space**. ``views`` = number of cameras used to solve the
|
|
point (1 ⇒ single-view ground-plane fallback, contract #5); ``quality`` is 0..1.
|
|
Points are always read ordered by ``t_global_s``.
|
|
"""
|
|
|
|
__tablename__ = "track_points"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
track_id: Mapped[int] = mapped_column(ForeignKey("tracks.id"), nullable=False, index=True)
|
|
t_global_s: Mapped[float] = mapped_column(Float, nullable=False)
|
|
x: Mapped[float] = mapped_column(Float, nullable=False)
|
|
y: Mapped[float] = mapped_column(Float, nullable=False)
|
|
z: Mapped[float] = mapped_column(Float, nullable=False)
|
|
quality: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
views: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Engine / session management
|
|
# ---------------------------------------------------------------------------
|
|
_engine: Engine | None = None
|
|
_SessionLocal: sessionmaker[Session] | None = None
|
|
_bound_path: Path | None = None
|
|
|
|
|
|
def init_engine(db_path: str | Path | None = None) -> Engine:
|
|
"""(Re)bind the module to a SQLite database file and return the engine.
|
|
|
|
Called with no argument uses ``config.DB_PATH``. Idempotent for the same path.
|
|
"""
|
|
global _engine, _SessionLocal, _bound_path
|
|
path = Path(db_path) if db_path is not None else config.DB_PATH
|
|
if _engine is not None and _bound_path == path:
|
|
return _engine
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if _engine is not None:
|
|
_engine.dispose()
|
|
_engine = create_engine(
|
|
f"sqlite:///{path}",
|
|
connect_args={"check_same_thread": False},
|
|
future=True,
|
|
)
|
|
_SessionLocal = sessionmaker(bind=_engine, expire_on_commit=False, future=True)
|
|
_bound_path = path
|
|
return _engine
|
|
|
|
|
|
def engine() -> Engine:
|
|
if _engine is None:
|
|
init_engine()
|
|
assert _engine is not None
|
|
return _engine
|
|
|
|
|
|
@contextmanager
|
|
def session_scope() -> Iterator[Session]:
|
|
"""Transactional session context; commits on success, rolls back on error."""
|
|
if _SessionLocal is None:
|
|
init_engine()
|
|
assert _SessionLocal is not None
|
|
session = _SessionLocal()
|
|
try:
|
|
yield session
|
|
session.commit()
|
|
except Exception:
|
|
session.rollback()
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def init_db() -> None:
|
|
"""Create all tables if they do not exist."""
|
|
Base.metadata.create_all(engine())
|
|
|
|
|
|
def reset_db() -> None:
|
|
"""Drop and recreate all tables (used by the synthetic fixture generator)."""
|
|
Base.metadata.drop_all(engine())
|
|
Base.metadata.create_all(engine())
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Videos
|
|
# ---------------------------------------------------------------------------
|
|
def add_video(
|
|
filename: str,
|
|
duration_s: float,
|
|
fps: float,
|
|
width: int,
|
|
height: int,
|
|
offset_ms: float | None = None,
|
|
drift_ppm: float | None = None,
|
|
sync_confidence: float | None = None,
|
|
) -> Video:
|
|
"""Insert a video row and return it (used by ingest + synthetic)."""
|
|
with session_scope() as s:
|
|
video = Video(
|
|
filename=filename,
|
|
duration_s=duration_s,
|
|
fps=fps,
|
|
width=width,
|
|
height=height,
|
|
offset_ms=offset_ms,
|
|
drift_ppm=drift_ppm,
|
|
sync_confidence=sync_confidence,
|
|
created_at=_now_iso(),
|
|
)
|
|
s.add(video)
|
|
s.flush()
|
|
s.refresh(video)
|
|
return video
|
|
|
|
|
|
def get_videos() -> list[Video]:
|
|
with session_scope() as s:
|
|
return list(s.scalars(select(Video).order_by(Video.id)))
|
|
|
|
|
|
def get_video(video_id: int) -> Video | None:
|
|
with session_scope() as s:
|
|
return s.get(Video, video_id)
|
|
|
|
|
|
def update_video_sync(
|
|
video_id: int,
|
|
offset_ms: float | None,
|
|
drift_ppm: float | None,
|
|
sync_confidence: float | None,
|
|
) -> None:
|
|
"""Persist the audio-sync solution for a video (lane A / M1)."""
|
|
with session_scope() as s:
|
|
video = s.get(Video, video_id)
|
|
if video is None:
|
|
raise KeyError(f"no video with id={video_id}")
|
|
video.offset_ms = offset_ms
|
|
video.drift_ppm = drift_ppm
|
|
video.sync_confidence = sync_confidence
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Camera poses
|
|
# ---------------------------------------------------------------------------
|
|
def set_poses(video_id: int, poses: Iterable[dict]) -> int:
|
|
"""Replace all poses for a video with ``poses`` (lane B / M2, and synthetic).
|
|
|
|
Each dict must have keys: frame_idx, t_video_s, qw, qx, qy, qz, tx, ty, tz,
|
|
fx, fy, cx, cy, and optionally ``registered`` (default True). Returns the count.
|
|
|
|
This is atomic: existing poses are only deleted inside the same transaction that
|
|
inserts the new ones, so a caller that raises before returning never corrupts the DB.
|
|
"""
|
|
rows = list(poses)
|
|
with session_scope() as s:
|
|
s.execute(delete(CameraPose).where(CameraPose.video_id == video_id))
|
|
for p in rows:
|
|
s.add(
|
|
CameraPose(
|
|
video_id=video_id,
|
|
frame_idx=int(p["frame_idx"]),
|
|
t_video_s=float(p["t_video_s"]),
|
|
qw=float(p["qw"]), qx=float(p["qx"]), qy=float(p["qy"]), qz=float(p["qz"]),
|
|
tx=float(p["tx"]), ty=float(p["ty"]), tz=float(p["tz"]),
|
|
fx=float(p["fx"]), fy=float(p["fy"]), cx=float(p["cx"]), cy=float(p["cy"]),
|
|
registered=bool(p.get("registered", True)),
|
|
)
|
|
)
|
|
return len(rows)
|
|
|
|
|
|
def get_poses(video_id: int) -> list[CameraPose]:
|
|
with session_scope() as s:
|
|
return list(
|
|
s.scalars(
|
|
select(CameraPose)
|
|
.where(CameraPose.video_id == video_id)
|
|
.order_by(CameraPose.t_video_s)
|
|
)
|
|
)
|
|
|
|
|
|
def has_poses() -> bool:
|
|
with session_scope() as s:
|
|
return s.scalar(select(CameraPose.id).limit(1)) is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Anchors
|
|
# ---------------------------------------------------------------------------
|
|
def add_anchor(label: str, x: float, y: float, z: float, color: str | None = None) -> Anchor:
|
|
with session_scope() as s:
|
|
anchor = Anchor(label=label, x=x, y=y, z=z, color=color)
|
|
s.add(anchor)
|
|
s.flush()
|
|
s.refresh(anchor)
|
|
return anchor
|
|
|
|
|
|
def get_anchors() -> list[Anchor]:
|
|
with session_scope() as s:
|
|
return list(s.scalars(select(Anchor).order_by(Anchor.id)))
|
|
|
|
|
|
def get_anchor(anchor_id: int) -> Anchor | None:
|
|
with session_scope() as s:
|
|
return s.get(Anchor, anchor_id)
|
|
|
|
|
|
def update_anchor(
|
|
anchor_id: int,
|
|
x: float,
|
|
y: float,
|
|
z: float,
|
|
label: str | None = None,
|
|
color: str | None = None,
|
|
) -> Anchor:
|
|
"""Move an anchor in place (M8 supersede). Position always updates; label/color only if given.
|
|
|
|
Raises ``KeyError`` if the id is unknown.
|
|
"""
|
|
with session_scope() as s:
|
|
anchor = s.get(Anchor, anchor_id)
|
|
if anchor is None:
|
|
raise KeyError(f"no anchor with id={anchor_id}")
|
|
anchor.x, anchor.y, anchor.z = x, y, z
|
|
if label is not None:
|
|
anchor.label = label
|
|
if color is not None:
|
|
anchor.color = color
|
|
s.flush()
|
|
s.refresh(anchor)
|
|
return anchor
|
|
|
|
|
|
def patch_anchor(anchor_id: int, label: str | None = None, color: str | None = None) -> Anchor:
|
|
"""Rename / recolor an anchor without moving it (phase 5 contract #4, M13).
|
|
|
|
Only provided fields change. Raises ``KeyError`` if the id is unknown.
|
|
"""
|
|
with session_scope() as s:
|
|
anchor = s.get(Anchor, anchor_id)
|
|
if anchor is None:
|
|
raise KeyError(f"no anchor with id={anchor_id}")
|
|
if label is not None:
|
|
anchor.label = label
|
|
if color is not None:
|
|
anchor.color = color
|
|
s.flush()
|
|
s.refresh(anchor)
|
|
return anchor
|
|
|
|
|
|
def delete_anchor(anchor_id: int) -> bool:
|
|
"""Delete an anchor, first unlinking any annotations that resolved to it.
|
|
|
|
Returns ``True`` if a row was deleted, ``False`` if the id was unknown. SQLite doesn't
|
|
enforce the FK by default, so we NULL ``annotations.resolved_anchor_id`` explicitly to keep
|
|
the link consistent.
|
|
"""
|
|
with session_scope() as s:
|
|
anchor = s.get(Anchor, anchor_id)
|
|
if anchor is None:
|
|
return False
|
|
s.execute(
|
|
update(Annotation)
|
|
.where(Annotation.resolved_anchor_id == anchor_id)
|
|
.values(resolved_anchor_id=None)
|
|
)
|
|
s.delete(anchor)
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Events
|
|
# ---------------------------------------------------------------------------
|
|
def add_event(
|
|
t_global_s: float,
|
|
event_type: str,
|
|
source: str,
|
|
duration_s: float | None = None,
|
|
confidence: float | None = None,
|
|
description: str | None = None,
|
|
) -> Event:
|
|
with session_scope() as s:
|
|
event = Event(
|
|
t_global_s=t_global_s,
|
|
duration_s=duration_s,
|
|
event_type=event_type,
|
|
confidence=confidence,
|
|
description=description,
|
|
source=source,
|
|
)
|
|
s.add(event)
|
|
s.flush()
|
|
s.refresh(event)
|
|
return event
|
|
|
|
|
|
def get_events() -> list[Event]:
|
|
with session_scope() as s:
|
|
return list(s.scalars(select(Event).order_by(Event.t_global_s)))
|
|
|
|
|
|
def clear_events(source: str | None = None) -> int:
|
|
"""Delete events, optionally only those from a given ``source``. Returns count deleted."""
|
|
with session_scope() as s:
|
|
stmt = delete(Event)
|
|
if source is not None:
|
|
stmt = stmt.where(Event.source == source)
|
|
result = s.execute(stmt)
|
|
return int(result.rowcount or 0)
|
|
|
|
|
|
def get_event(event_id: int) -> Event | None:
|
|
with session_scope() as s:
|
|
return s.get(Event, event_id)
|
|
|
|
|
|
def update_event(
|
|
event_id: int,
|
|
event_type: str | None = None,
|
|
source: str | None = None,
|
|
description: str | None = None,
|
|
confidence: float | None = None,
|
|
) -> Event:
|
|
"""Patch an existing event in place (M8 correction). Only provided fields change.
|
|
|
|
Returns the updated row; raises ``KeyError`` if the id is unknown.
|
|
"""
|
|
with session_scope() as s:
|
|
event = s.get(Event, event_id)
|
|
if event is None:
|
|
raise KeyError(f"no event with id={event_id}")
|
|
if event_type is not None:
|
|
event.event_type = event_type
|
|
if source is not None:
|
|
event.source = source
|
|
if description is not None:
|
|
event.description = description
|
|
if confidence is not None:
|
|
event.confidence = confidence
|
|
s.flush()
|
|
s.refresh(event)
|
|
return event
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Saved camera paths (phase 5 / M16)
|
|
# ---------------------------------------------------------------------------
|
|
def add_path(name: str, path_json: str) -> SavedPath:
|
|
with session_scope() as s:
|
|
path = SavedPath(name=name, path_json=path_json, created_at=_now_iso())
|
|
s.add(path)
|
|
s.flush()
|
|
s.refresh(path)
|
|
return path
|
|
|
|
|
|
def get_paths() -> list[SavedPath]:
|
|
with session_scope() as s:
|
|
return list(s.scalars(select(SavedPath).order_by(SavedPath.id)))
|
|
|
|
|
|
def get_path(path_id: int) -> SavedPath | None:
|
|
with session_scope() as s:
|
|
return s.get(SavedPath, path_id)
|
|
|
|
|
|
def delete_path(path_id: int) -> bool:
|
|
"""Delete a saved path. Returns ``True`` if a row was deleted, ``False`` if unknown."""
|
|
with session_scope() as s:
|
|
path = s.get(SavedPath, path_id)
|
|
if path is None:
|
|
return False
|
|
s.delete(path)
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Friend tracks (phase 6 / M18 contract #2). Written by the solver (lane H / M20),
|
|
# read by the API + frontend. TrackPoints are always returned ordered by t_global_s.
|
|
# ---------------------------------------------------------------------------
|
|
def add_track(marker_key: str, label: str | None = None, color: str | None = None) -> Track:
|
|
"""Insert a track row (no points yet) and return it."""
|
|
with session_scope() as s:
|
|
track = Track(marker_key=marker_key, label=label, color=color, created_at=_now_iso())
|
|
s.add(track)
|
|
s.flush()
|
|
s.refresh(track)
|
|
return track
|
|
|
|
|
|
def get_tracks() -> list[Track]:
|
|
"""All tracks, ordered by id."""
|
|
with session_scope() as s:
|
|
return list(s.scalars(select(Track).order_by(Track.id)))
|
|
|
|
|
|
def get_track(track_id: int) -> Track | None:
|
|
with session_scope() as s:
|
|
return s.get(Track, track_id)
|
|
|
|
|
|
def get_track_points(track_id: int) -> list[TrackPoint]:
|
|
"""A track's points, ordered by ``t_global_s`` (contract #2)."""
|
|
with session_scope() as s:
|
|
return list(
|
|
s.scalars(
|
|
select(TrackPoint)
|
|
.where(TrackPoint.track_id == track_id)
|
|
.order_by(TrackPoint.t_global_s)
|
|
)
|
|
)
|
|
|
|
|
|
def set_track_points(track_id: int, points: Iterable[dict]) -> int:
|
|
"""Replace all points for a track with ``points`` (solver output, M20). Returns the count.
|
|
|
|
Each dict needs ``t_global_s, x, y, z`` and optionally ``quality`` (default None) and
|
|
``views`` (default None). Atomic: the delete + inserts share one transaction, so a caller
|
|
that raises mid-way never leaves a half-written track. Raises ``KeyError`` if the track
|
|
is unknown.
|
|
"""
|
|
rows = list(points)
|
|
with session_scope() as s:
|
|
if s.get(Track, track_id) is None:
|
|
raise KeyError(f"no track with id={track_id}")
|
|
s.execute(delete(TrackPoint).where(TrackPoint.track_id == track_id))
|
|
for p in rows:
|
|
s.add(
|
|
TrackPoint(
|
|
track_id=track_id,
|
|
t_global_s=float(p["t_global_s"]),
|
|
x=float(p["x"]), y=float(p["y"]), z=float(p["z"]),
|
|
quality=None if p.get("quality") is None else float(p["quality"]),
|
|
views=None if p.get("views") is None else int(p["views"]),
|
|
)
|
|
)
|
|
return len(rows)
|
|
|
|
|
|
def patch_track(track_id: int, label: str | None = None, color: str | None = None) -> Track:
|
|
"""Rename / recolor a track (M21 panel). Only provided fields change.
|
|
|
|
Raises ``KeyError`` if the id is unknown.
|
|
"""
|
|
with session_scope() as s:
|
|
track = s.get(Track, track_id)
|
|
if track is None:
|
|
raise KeyError(f"no track with id={track_id}")
|
|
if label is not None:
|
|
track.label = label
|
|
if color is not None:
|
|
track.color = color
|
|
s.flush()
|
|
s.refresh(track)
|
|
return track
|
|
|
|
|
|
def delete_track(track_id: int) -> bool:
|
|
"""Delete a track and its points. Returns ``True`` if a row was deleted, else ``False``.
|
|
|
|
SQLite doesn't cascade by default, so the points are removed explicitly.
|
|
"""
|
|
with session_scope() as s:
|
|
track = s.get(Track, track_id)
|
|
if track is None:
|
|
return False
|
|
s.execute(delete(TrackPoint).where(TrackPoint.track_id == track_id))
|
|
s.delete(track)
|
|
return True
|
|
|
|
|
|
def has_tracks() -> bool:
|
|
"""Whether any track exists (drives ``manifest.has_tracks``)."""
|
|
with session_scope() as s:
|
|
return s.scalar(select(Track.id).limit(1)) is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Annotations
|
|
# ---------------------------------------------------------------------------
|
|
def add_annotation(
|
|
video_id: int,
|
|
t_video_s: float,
|
|
x0: float,
|
|
y0: float,
|
|
x1: float,
|
|
y1: float,
|
|
event_id: int | None = None,
|
|
resolved_anchor_id: int | None = None,
|
|
) -> Annotation:
|
|
with session_scope() as s:
|
|
annotation = Annotation(
|
|
video_id=video_id,
|
|
t_video_s=t_video_s,
|
|
x0=x0, y0=y0, x1=x1, y1=y1,
|
|
event_id=event_id,
|
|
resolved_anchor_id=resolved_anchor_id,
|
|
)
|
|
s.add(annotation)
|
|
s.flush()
|
|
s.refresh(annotation)
|
|
return annotation
|
|
|
|
|
|
def get_annotations(
|
|
video_id: int | None = None, event_id: int | None = None
|
|
) -> list[Annotation]:
|
|
with session_scope() as s:
|
|
stmt = select(Annotation).order_by(Annotation.id)
|
|
if video_id is not None:
|
|
stmt = stmt.where(Annotation.video_id == video_id)
|
|
if event_id is not None:
|
|
stmt = stmt.where(Annotation.event_id == event_id)
|
|
return list(s.scalars(stmt))
|
|
|
|
|
|
def set_annotation_anchor(annotation_id: int, anchor_id: int) -> None:
|
|
with session_scope() as s:
|
|
annotation = s.get(Annotation, annotation_id)
|
|
if annotation is None:
|
|
raise KeyError(f"no annotation with id={annotation_id}")
|
|
annotation.resolved_anchor_id = anchor_id
|