feat(trip-cluster): coverage detection — completeness flags + outliers

apps/trip-cluster/app/coverage.py

@@ -0,0 +1,35 @@
+from photoflow.core.models import ClusterMember
+from app.clustering import _epoch, _median
+
+
+def coverage_members(candidate, assets_by_id, *, outlier_factor: float = 8.0) -> list:
+    member_ids = [i for i in candidate.member_ids if i in assets_by_id]
+    members = sorted((assets_by_id[i] for i in member_ids), key=lambda a: a.taken_at)
+    epochs = [_epoch(a.taken_at) for a in members]
+    gaps = [b - a for a, b in zip(epochs, epochs[1:])]
+    base = _median(gaps) if gaps else 0.0
+
+    out = []
+    for idx, a in enumerate(members):
+        is_outlier = False
+        if candidate.seed_tag and base > 0:
+            left = epochs[idx] - epochs[idx - 1] if idx > 0 else 0
+            right = epochs[idx + 1] - epochs[idx] if idx < len(members) - 1 else 0
+            nearest = min([g for g in (left, right) if g > 0], default=0)
+            if nearest > outlier_factor * base:
+                is_outlier = True
+        out.append(ClusterMember(
+            cluster_id=0, immich_id=a.immich_id,
+            member_confidence=candidate.confidence, is_outlier=is_outlier,
+            included=True, flagged_coverage=False))
+
+    if candidate.seed_tag and candidate.start_at and candidate.end_at:
+        mset = set(member_ids)
+        for a in sorted(assets_by_id.values(), key=lambda x: x.taken_at):
+            if a.immich_id in mset:
+                continue
+            if candidate.start_at <= a.taken_at <= candidate.end_at:
+                out.append(ClusterMember(
+                    cluster_id=0, immich_id=a.immich_id, member_confidence=0.0,
+                    is_outlier=False, included=False, flagged_coverage=True))
+    return out

apps/trip-cluster/tests/test_coverage.py

@@ -0,0 +1,48 @@
+from photoflow.core.models import Asset
+from app.clustering import CandidateCluster
+from app.coverage import coverage_members
+
+
+def _a(i, taken):
+    return Asset(immich_id=i, taken_at=taken)
+
+
+def test_coverage_candidate_inside_seeded_window():
+    members = [_a("a", "2019-06-01T10:00:00"), _a("b", "2019-06-01T12:00:00"),
+               _a("c", "2019-06-02T10:00:00")]
+    intruder = _a("x", "2019-06-01T13:00:00")          # in window, untagged
+    outside = _a("y", "2019-07-01T10:00:00")           # out of window
+    by_id = {m.immich_id: m for m in members + [intruder, outside]}
+    cand = CandidateCluster(member_ids=["a", "b", "c"], start_at="2019-06-01T10:00:00",
+                            end_at="2019-06-02T10:00:00", suggested_name="Italy 2019",
+                            confidence=0.95, kind_guess="trip", seed_tag="Italy 2019")
+    out = coverage_members(cand, by_id)
+    flagged = {m.immich_id for m in out if m.flagged_coverage}
+    assert flagged == {"x"}                            # only the in-window intruder
+    assert all(not m.included for m in out if m.flagged_coverage)
+
+
+def test_outlier_member_far_from_bulk():
+    members = [_a("a", "2019-06-01T10:00:00"), _a("a2", "2019-06-01T11:00:00"),
+               _a("a3", "2019-06-01T12:00:00"),
+               _a("z", "2019-09-01T10:00:00")]          # tagged but months away
+    by_id = {m.immich_id: m for m in members}
+    cand = CandidateCluster(member_ids=["a", "a2", "a3", "z"],
+                            start_at="2019-06-01T10:00:00", end_at="2019-09-01T10:00:00",
+                            suggested_name="Italy 2019", confidence=0.95,
+                            kind_guess="trip", seed_tag="Italy 2019")
+    out = coverage_members(cand, by_id)
+    outliers = {m.immich_id for m in out if m.is_outlier}
+    assert outliers == {"z"}
+
+
+def test_non_seeded_cluster_has_no_coverage_or_outliers():
+    members = [_a("a", "2019-06-01T10:00:00"), _a("b", "2019-06-01T12:00:00")]
+    intruder = _a("x", "2019-06-01T11:00:00")
+    by_id = {m.immich_id: m for m in members + [intruder]}
+    cand = CandidateCluster(member_ids=["a", "b"], start_at="2019-06-01T10:00:00",
+                            end_at="2019-06-01T12:00:00", suggested_name="Trip",
+                            confidence=0.4, kind_guess="everyday", seed_tag=None)
+    out = coverage_members(cand, by_id)
+    assert {m.immich_id for m in out} == {"a", "b"}
+    assert not any(m.flagged_coverage or m.is_outlier for m in out)