from photoflow.core.models import Asset from app.clustering import cluster_assets def _a(i, taken, gps=False, city=None): return Asset(immich_id=i, taken_at=taken, gps_lat=45.0 if gps else None, gps_lon=12.0 if gps else None, place_city=city) def test_seed_tag_forms_one_cluster_not_gap_split(): # Two assets months apart but sharing a trip tag -> ONE seeded cluster. assets = [_a("a", "2019-06-01T10:00:00"), _a("b", "2019-09-01T10:00:00")] tags = {"a": ["Italy 2019"], "b": ["Italy 2019"]} clusters = cluster_assets(assets, tags, {"Italy 2019"}) assert len(clusters) == 1 c = clusters[0] assert c.seed_tag == "Italy 2019" and sorted(c.member_ids) == ["a", "b"] assert c.confidence >= 0.9 and c.suggested_name == "Italy 2019" def test_sparse_old_regime_splits_on_adaptive_threshold(): # ~1 day intra-trip gaps; trips separated by 10 days (< 14d hard cap), # so only the adaptive rule can split them. a = [_a(f"a{i}", f"2008-06-0{i+1}T12:00:00") for i in range(5)] # Jun 1..5 b = [_a(f"b{i}", f"2008-06-1{i+5}T12:00:00") for i in range(3)] # Jun 15..17 assets = a + b tags = {x.immich_id: [] for x in assets} clusters = cluster_assets(assets, tags, set()) assert len(clusters) == 2 assert sorted(clusters[0].member_ids) == ["a0", "a1", "a2", "a3", "a4"] def test_dense_recent_regime_splits_on_adaptive_threshold(): # Hourly bursts within a day; 2-day gap between days. day1 = [_a(f"d{i}", f"2024-03-10T{10+i:02d}:00:00") for i in range(4)] day3 = [_a(f"e{i}", f"2024-03-12T{10+i:02d}:00:00") for i in range(4)] assets = day1 + day3 tags = {x.immich_id: [] for x in assets} clusters = cluster_assets(assets, tags, set()) assert len(clusters) == 2 assert sorted(clusters[0].member_ids) == ["d0", "d1", "d2", "d3"] def test_location_anchor_names_and_gps_confidence(): assets = [_a("a", "2020-05-01T10:00:00", gps=True, city="Kiev"), _a("b", "2020-05-01T12:00:00", gps=True, city="Kiev"), _a("c", "2020-05-01T14:00:00", gps=True, city="Kiev"), _a("d", "2020-05-01T16:00:00", gps=True, city="Kiev"), _a("e", "2020-05-01T18:00:00", gps=True, city="Kiev")] tags = {x.immich_id: [] for x in assets} c = cluster_assets(assets, tags, set())[0] assert c.suggested_name == "Kiev" assert c.confidence > 0.6 # full GPS lifts confidence assert c.kind_guess == "trip" def test_small_scattered_cluster_marked_everyday(): assets = [_a("a", "2015-01-01T10:00:00"), _a("b", "2015-01-01T11:00:00")] tags = {"a": [], "b": []} c = cluster_assets(assets, tags, set())[0] assert c.kind_guess == "everyday"