From 5256df4117b4ef76fdc36fd16e4d8d00b6ab3065 Mon Sep 17 00:00:00 2001 From: mischa Date: Sat, 27 Jun 2026 17:28:27 +0200 Subject: [PATCH] feat(trip-cluster): cluster orchestration (seed derivation + coverage) + CLI --- apps/trip-cluster/app/cli.py | 14 ++++- apps/trip-cluster/app/cluster_run.py | 57 +++++++++++++++++++ apps/trip-cluster/tests/test_cluster_run.py | 61 +++++++++++++++++++++ 3 files changed, 131 insertions(+), 1 deletion(-) create mode 100644 apps/trip-cluster/app/cluster_run.py create mode 100644 apps/trip-cluster/tests/test_cluster_run.py diff --git a/apps/trip-cluster/app/cli.py b/apps/trip-cluster/app/cli.py index 956a55d..4bb7023 100644 --- a/apps/trip-cluster/app/cli.py +++ b/apps/trip-cluster/app/cli.py @@ -31,6 +31,16 @@ def cmd_ingest(deps, *, date_from, date_to, tag, subset, full) -> int: return 0 +def cmd_cluster(deps, *, gap_factor) -> int: + cfg = deps["config"] + store = _store(cfg) + from app.cluster_run import run_cluster + res = run_cluster(store, gap_factor=gap_factor) + store.close() + print(f"Built {res['clusters']} candidate cluster(s). Run 'categorize serve' to review.") + return 0 + + def cmd_serve(deps) -> int: from app import create_app create_app(deps["config"]).run(host="0.0.0.0", port=8084) @@ -67,7 +77,9 @@ def main(argv=None) -> int: if args.command == "ingest": return cmd_ingest(deps, date_from=args.date_from, date_to=args.date_to, tag=args.tag, subset=args.subset, full=args.full) - # cluster / apply are wired in later tasks. + if args.command == "cluster": + return cmd_cluster(deps, gap_factor=args.gap_factor) + # apply is wired in a later task. print(f"Command '{args.command}' is not implemented yet.") return 1 diff --git a/apps/trip-cluster/app/cluster_run.py b/apps/trip-cluster/app/cluster_run.py new file mode 100644 index 0000000..eb58ea6 --- /dev/null +++ b/apps/trip-cluster/app/cluster_run.py @@ -0,0 +1,57 @@ +from collections import defaultdict + +from photoflow.core.models import Cluster +from photoflow.immich import pipeline +from app.clustering import cluster_assets, _epoch +from app.coverage import coverage_members + +# A tag seeds only if its photos dominate the time window they span: among all +# assets whose taken_at falls in [min, max] of the tag's photos, at least this +# fraction must carry the tag. A name/event tag sprinkled among many untagged +# photos in the same window fails this gate. +SEED_DOMINANCE_MIN = 0.5 + + +def _seed_tags(assets, tags_by_asset, max_span_days: int, + dominance_min: float = SEED_DOMINANCE_MIN) -> set: + epochs = {a.immich_id: _epoch(a.taken_at) for a in assets} + tagged = defaultdict(list) # tag -> epochs of its assets + for a in assets: + for t in tags_by_asset.get(a.immich_id, []): + if not pipeline.is_pipeline_tag(t): + tagged[t].append(epochs[a.immich_id]) + + seeds = set() + for tag, ts in tagged.items(): + if len(ts) < 2: + continue + lo, hi = min(ts), max(ts) + # Loose safety cap only — reject absurdly long spans, but keep genuine + # long trips (this library has one > 60 days), so the default is high. + if hi - lo > max_span_days * 86400: + continue + # Dominance / temporal contiguity gate. + in_window = sum(1 for e in epochs.values() if lo <= e <= hi) + if in_window and len(ts) / in_window >= dominance_min: + seeds.add(tag) + return seeds + + +def run_cluster(store, *, gap_factor: float = 6.0, seed_max_span_days: int = 400) -> dict: + assets = store.all_assets(include_processed=False) + tags_by_asset = {a.immich_id: store.asset_tags(a.immich_id) for a in assets} + seed_tags = _seed_tags(assets, tags_by_asset, seed_max_span_days) + candidates = cluster_assets(assets, tags_by_asset, seed_tags, gap_factor=gap_factor) + assets_by_id = {a.immich_id: a for a in assets} + + store.clear_clusters() + n = 0 + for cand in candidates: + members = coverage_members(cand, assets_by_id) + store.insert_cluster(Cluster( + start_at=cand.start_at, end_at=cand.end_at, + count=sum(1 for m in members if m.included), + suggested_name=cand.suggested_name, confidence=cand.confidence, + kind_guess=cand.kind_guess, status="pending"), members) + n += 1 + return {"clusters": n} diff --git a/apps/trip-cluster/tests/test_cluster_run.py b/apps/trip-cluster/tests/test_cluster_run.py new file mode 100644 index 0000000..ccc38db --- /dev/null +++ b/apps/trip-cluster/tests/test_cluster_run.py @@ -0,0 +1,61 @@ +from photoflow.core import Store +from photoflow.core.models import Asset +from photoflow.immich import pipeline +from app.cluster_run import run_cluster + + +def _store(tmp_path): + return Store(str(tmp_path / "t.db")).connect() + + +def test_run_cluster_seeds_bounded_tag_not_people_tag(tmp_path): + s = _store(tmp_path) + # "Italy 2019" spans 2 days -> seeds; "Mum" spans years -> not a seed. + rows = [("a", "2019-06-01T10:00:00", ["Italy 2019", "Mum"]), + ("b", "2019-06-02T10:00:00", ["Italy 2019"]), + ("c", "2010-01-01T10:00:00", ["Mum"]), + ("d", "2022-01-01T10:00:00", ["Mum"])] + for i, t, tags in rows: + s.upsert_asset(Asset(immich_id=i, taken_at=t)) + s.set_asset_tags(i, tags) + res = run_cluster(s) + seeded = [c for c in s.all_clusters() if c.suggested_name == "Italy 2019"] + assert len(seeded) == 1 + assert sorted(x.immich_id for x, _ in s.cluster_members(seeded[0].id)) == ["a", "b"] + assert res["clusters"] >= 1 + s.close() + + +def test_run_cluster_excludes_processed_and_replaces(tmp_path): + s = _store(tmp_path) + s.upsert_asset(Asset(immich_id="a", taken_at="2019-06-01T10:00:00")) + s.upsert_asset(Asset(immich_id="p", taken_at="2019-06-01T11:00:00")) + s.set_asset_tags("a", []) + s.set_asset_tags("p", [pipeline.PROCESSED]) + s.mark_processed("p") + run_cluster(s) + run_cluster(s) # idempotent replace — not doubled + all_member_ids = [x.immich_id for c in s.all_clusters() + for x, _ in s.cluster_members(c.id)] + assert "p" not in all_member_ids + assert all_member_ids.count("a") == 1 + s.close() + + +def test_run_cluster_short_span_non_dominant_tag_does_not_seed(tmp_path): + s = _store(tmp_path) + # "Lunch" is on 2 photos within a ~30-min window, but that same window is + # full of untagged photos -> the tag does NOT dominate its window + # (2/5 = 0.4 < 0.5), so it must not seed even though its span is tiny. + rows = [("x", "2020-03-01T12:00:00", ["Lunch"]), + ("y", "2020-03-01T12:30:00", ["Lunch"]), + ("u1", "2020-03-01T12:05:00", []), + ("u2", "2020-03-01T12:15:00", []), + ("u3", "2020-03-01T12:25:00", [])] + for i, t, tags in rows: + s.upsert_asset(Asset(immich_id=i, taken_at=t)) + s.set_asset_tags(i, tags) + run_cluster(s) + seeded = [c for c in s.all_clusters() if c.suggested_name == "Lunch"] + assert seeded == [] + s.close()