#!/usr/bin/env python3 """Generate db_schema.svg — an entity-relationship diagram of the alert4k data model. Hand-curated layout. Re-run to regenerate the SVG after schema changes. """ # ---- visual constants ------------------------------------------------------- HEADER_H = 28 ROW_H = 17 PAD_X = 10 FONT = "font-family='Menlo, Consolas, monospace'" # table accent colours C_HUB = "#2c3e7a" # events C_DET = "#1f7a4d" # detection-ish (detections, clip_detections, clip_tracks) C_ID = "#9a4d1f" # identity (object_identities) C_REF = "#6a3d8a" # reference dictionaries (known/tracked objects) C_MOT = "#7a1f3d" # motion / scene infra # (label, kind) kind: 'pk','fk','col' def col(label, kind="col"): return (label, kind) tables = { "events": { "x": 470, "y": 90, "w": 270, "accent": C_HUB, "subtitle": "ROOT — one row per motion event", "cols": [ col("id", "pk"), col("trigger_ts timestamptz"), col("source_name varchar"), col("motion_score float8"), col("snapshot_path text"), col("clip_path text"), col("frame4k_path text"), col("annotated_frame_path text"), col("description text"), col("false_positive bool"), col("absorbed / san_suppress bool"), col("clip_duration_seconds f4"), col("motion_roi_x1..y2 float"), col("blob_count/total/min/max"), col("pipeline_stage varchar"), col("clip_tracks_version int"), col("scene_background_id → scene_backgrounds", "fk"), col("created_at timestamptz"), ], }, "scene_backgrounds": { "x": 70, "y": 70, "w": 250, "accent": C_MOT, "subtitle": "camera calibration backdrop", "cols": [ col("id", "pk"), col("camera_id varchar"), col("image_path text"), col("scene_version_id int"), col("quality_score float"), col("pose_signature text"), col("created_at timestamptz"), ], }, "motion_detections": { "x": 70, "y": 420, "w": 250, "accent": C_MOT, "subtitle": "raw MOG2 blobs (motion svc)", "cols": [ col("id", "pk"), col("event_id → events (NULL=unclaimed)", "fk"), col("frame_ts timestamptz"), col("blob_x1..y2 float"), col("area_px int"), col("motion_ratio float"), ], }, "detections": { "x": 470, "y": 560, "w": 270, "accent": C_DET, "subtitle": "single-frame GDINO @ trigger", "cols": [ col("id", "pk"), col("event_id → events", "fk"), col("phrase varchar"), col("score float8"), col("cx, cy, w, h (normalized)"), col("created_at"), ], }, "clip_tracks": { "x": 840, "y": 70, "w": 280, "accent": C_DET, "subtitle": "one row per ByteTrack track in clip", "cols": [ col("id", "pk"), col("event_id → events", "fk"), col("track_id int"), col("phrase varchar"), col("first_seen_s / last_seen_s"), col("frame_count int"), col("max_score float"), col("path_json text [{t,cx,cy}]"), col("tags jsonb"), col("displacement float"), col("path_metrics real[]"), col("quality_metrics real[]"), col("track_embedding bytea (CLIP)"), col("image_quality / caption_quality"), col("tq_suppress bool"), col("alpha_crop_path text"), ], }, "clip_detections": { "x": 840, "y": 540, "w": 280, "accent": C_DET, "subtitle": "per-frame GDINO sampled from clip", "cols": [ col("event_id → events", "fk"), col("track_id → clip_tracks (soft)", "fk"), col("frame_time_s real"), col("phrase char(1) code"), col("score real"), col("cx, cy, w, h real"), col("created_at"), col("(no PK since V15 — narrow table)"), ], }, "object_identities": { "x": 1230, "y": 360, "w": 290, "accent": C_ID, "subtitle": "JUNCTION — recognition result", "cols": [ col("id", "pk"), col("event_id → events", "fk"), col("detection_idx int"), col("phrase varchar"), col("embedding bytea (CLIP crop)"), col("crop_path text"), col("known_object_id → known_objects", "fk"), col("tracked_object_id → tracked_objects", "fk"), col("clip_track_id → clip_tracks", "fk"), col("identity_type 'detection'|'track'"), col("match_score float8"), col("is_new bool"), col("created_at"), ], }, "known_objects": { "x": 1230, "y": 70, "w": 290, "accent": C_REF, "subtitle": "manually registered named objects", "cols": [ col("id", "pk"), col("name varchar (e.g. 'my Honda')"), col("phrase varchar"), col("embedding bytea (CLIP 512f)"), col("thumbnail_path text"), col("notes text"), col("created_at"), ], }, "tracked_objects": { "x": 1230, "y": 720, "w": 290, "accent": C_REF, "subtitle": "auto-discovered, clustered cross-event", "cols": [ col("id", "pk"), col("pseudo_name varchar UNIQUE"), col("phrase varchar"), col("embedding bytea (EMA-updated)"), col("occurrence_count int"), col("first_seen_ts / last_seen_ts"), col("ground_x/y, world_x_m/y_m"), col("sfm_confidence / confidence_score"), col("identity_status text"), col("best_crop_path / thumbnail_path"), col("created_at"), ], }, } def table_height(t): return HEADER_H + 14 + ROW_H * len(t["cols"]) + 8 # anchor helpers — return (x,y) on the edge of a table box def left(t, row=None): y = t["y"] + HEADER_H + 14 + (ROW_H * row + ROW_H // 2 if row is not None else table_height(t)//2) return (t["x"], y) def right(t, row=None): y = t["y"] + HEADER_H + 14 + (ROW_H * row + ROW_H // 2 if row is not None else table_height(t)//2) return (t["x"] + t["w"], y) def top(t): return (t["x"] + t["w"]//2, t["y"]) def bottom(t): return (t["x"] + t["w"]//2, t["y"] + table_height(t)) parts = [] W, H = 1580, 980 parts.append(f"") parts.append(f"") # title parts.append(f"" "alert4k — Relational Data Model (PostgreSQL)") parts.append(f"" "Motion event → clip analysis → per-track CLIP identity → cross-event object clustering. " "Solid = enforced FK · Dashed = soft/logical link.") # ---- relationship edges (drawn first, under boxes) ------------------------- def _glyph(x, y, dir_x, sym): """Draw a cardinality glyph at endpoint (x,y). dir_x = +1 if the line extends to the right of the point, -1 if to the left. sym in {'one','many'}.""" o = 11 * (1 if dir_x >= 0 else -1) # offset into the line if sym == "one": bx = x + o * 0.55 parts.append(f"") else: # many — crow's foot fanning into (x,y) parts.append(f"") # (from_anchor, to_anchor, label, dashed) def edge(p1, p2, label, dashed=False, lx=None, ly=None, start="one", end="many"): dash = " stroke-dasharray='6 4'" if dashed else "" midx = (p1[0] + p2[0]) / 2 d = f"M{p1[0]},{p1[1]} L{midx},{p1[1]} L{midx},{p2[1]} L{p2[0]},{p2[1]}" parts.append(f"") _glyph(p1[0], p1[1], 1 if midx >= p1[0] else -1, start) _glyph(p2[0], p2[1], 1 if midx >= p2[0] else -1, end) if lx is None: lx = (p1[0] + p2[0]) / 2 if ly is None: ly = (p1[1] + p2[1]) / 2 parts.append(f"") parts.append(f"{label}") T = tables # events -> children (1:N). events.id is the 'one' side. edge(left(T["events"], 1), right(T["motion_detections"], 1), "1 : N", dashed=False) edge(bottom(T["events"]), top(T["detections"]), "1 : N") edge(right(T["events"], 1), left(T["clip_tracks"], 1), "1 : N") edge(right(T["events"], 5), left(T["clip_detections"], 0), "1 : N") edge(right(T["events"], 9), left(T["object_identities"], 1), "1 : N") # events.scene_background_id -> scene_backgrounds (soft, N:1 from events side) edge(left(T["events"], 16), right(T["scene_backgrounds"], 1), "N : 1", dashed=True, start="many", end="one") # clip_detections.track_id -> clip_tracks (soft composite) edge(top(T["clip_detections"]), bottom(T["clip_tracks"]), "N : 1", dashed=True, start="many", end="one") # object_identities FKs (N:1 to the reference tables) edge(top(T["object_identities"]), bottom(T["known_objects"]), "N : 1", start="many", end="one") edge(bottom(T["object_identities"]), top(T["tracked_objects"]), "N : 1", start="many", end="one") edge(left(T["object_identities"], 8), right(T["clip_tracks"], 7), "N : 1", start="many", end="one") # ---- table boxes ----------------------------------------------------------- def kind_color(kind): return {"pk": "#1a1a2e", "fk": "#9a4d1f", "col": "#333"}[kind] for name, t in tables.items(): h = table_height(t) x, y, w = t["x"], t["y"], t["w"] parts.append(f"") parts.append(f"") parts.append(f"") parts.append(f"{name}") parts.append(f"{t['subtitle']}") cy = y + HEADER_H + 14 for i, (label, kind) in enumerate(t["cols"]): ry = cy + ROW_H * i if i % 2 == 1: parts.append(f"") weight = "700" if kind == "pk" else "400" prefix = "PK " if kind == "pk" else ("FK " if kind == "fk" else " ") deco = " text-decoration='underline'" if kind == "pk" else "" esc = label.replace("&", "&").replace("<", "<").replace(">", "→") parts.append(f"" f"{prefix}{esc}") # ---- legend ---------------------------------------------------------------- lx, ly = 70, 720 parts.append(f"") parts.append(f"Legend") legend = [ ("PK primary key (underlined)", "#1a1a2e"), ("FK foreign key → target", "#9a4d1f"), ("─── enforced FK constraint", "#444"), ("- - soft / logical link", "#444"), ("⊢ = 'one' ⟨ = 'many'", "#444"), ("embedding = CLIP ViT-B/32 512f", "#666"), ] for i, (txt, c) in enumerate(legend): parts.append(f"{txt}") parts.append("") with open("db_schema.svg", "w") as f: f.write("\n".join(parts)) print("wrote db_schema.svg")