import datetime from collections import Counter from dataclasses import dataclass from typing import Optional EVERYDAY_MAX_COUNT = 4 SEED_CONFIDENCE = 0.95 # Public aliases for cross-module use are defined at the bottom of this file # (epoch / median) so a later coverage.py can consume them without reaching # for the underscore-prefixed names. The underscore names are kept too, since # Task 11 (coverage.py) imports `_epoch` / `_median` directly. __all__ = [ "CandidateCluster", "cluster_assets", "_epoch", "_median", "epoch", "median", ] @dataclass class CandidateCluster: member_ids: list start_at: str end_at: str suggested_name: str confidence: float kind_guess: str seed_tag: Optional[str] = None def _epoch(taken_at: str) -> float: s = (taken_at or "").strip() if not s: return 0.0 s = s.replace("Z", "") if "." in s: s = s.split(".", 1)[0] try: if "T" in s: return datetime.datetime.fromisoformat(s).timestamp() return datetime.datetime.fromisoformat(s + "T00:00:00").timestamp() except ValueError: return 0.0 def _median(values: list) -> float: if not values: return 0.0 xs = sorted(values) n = len(xs) mid = n // 2 return xs[mid] if n % 2 else (xs[mid - 1] + xs[mid]) / 2 def _span(members: list) -> tuple: ts = [m.taken_at for m in members] return (min(ts), max(ts)) if ts else ("", "") def _name(members: list, start_at: str) -> str: cities = Counter(m.place_city for m in members if m.place_city) if cities: return cities.most_common(1)[0][0] countries = Counter(m.place_country for m in members if m.place_country) if countries: return countries.most_common(1)[0][0] return f"Trip {start_at[:10]}" def _tightness(members: list) -> float: """Temporal-tightness score in [0, 1]: how regular the intra-cluster time gaps are (low gap variance -> high score). A densely/regularly shot cluster (e.g. a steady stream of photos through a day) is strong evidence of a coherent event even when GPS is absent. We measure regularity via the coefficient of variation (stdev / mean) of the consecutive gaps and reward a low value. This lets a GPS-poor but tightly packed cluster clear the downstream 0.75 bulk-approve gate, which the pure GPS+size score could never reach (it caps at 0.50 when gps_frac == 0). """ ts = sorted(_epoch(m.taken_at) for m in members) gaps = [b - a for a, b in zip(ts, ts[1:])] if not gaps: return 0.0 # single member: no temporal signal if len(gaps) < 2: return 1.0 # one gap: trivially regular mean = sum(gaps) / len(gaps) if mean <= 0: return 1.0 # all timestamps coincide: maximally tight var = sum((g - mean) ** 2 for g in gaps) / len(gaps) cv = (var ** 0.5) / mean return max(0.0, 1.0 - cv) def _confidence(members: list) -> float: """Heuristic confidence in [0, 0.85] that a free cluster is a real event. Term rationale: - 0.30 base: even a bare timestamp cluster is a weak positive signal, so we never start from zero. - 0.40 * gps_frac: geotagging is the strongest single signal that photos belong to one outing, hence the largest weight. - 0.20 * size_frac: more photos (saturating at 20) make a stray-photo false positive less likely. - 0.35 * tightness: regular/dense timing is independent evidence of a coherent event; weighted so a fully GPS-poor cluster can still reach the 0.85 cap (0.30 + 0.20 + 0.35) and clear the 0.75 approve gate. The 0.85 cap reserves >0.90 confidence exclusively for tag-seeded clusters. """ count = len(members) gps_frac = sum(1 for m in members if m.gps_lat is not None) / count if count else 0 size_frac = min(count / 20, 1) conf = 0.30 + 0.40 * gps_frac + 0.20 * size_frac + 0.35 * _tightness(members) return round(min(conf, 0.85), 2) def _free_cluster(members: list) -> CandidateCluster: start, end = _span(members) return CandidateCluster( member_ids=[m.immich_id for m in members], start_at=start, end_at=end, suggested_name=_name(members, start), confidence=_confidence(members), kind_guess="everyday" if len(members) <= EVERYDAY_MAX_COUNT else "trip") def _gap_cluster(assets: list, *, gap_factor, hard_split_days, min_floor_seconds) -> list: ordered = sorted(assets, key=lambda a: a.taken_at) if not ordered: return [] hard_cap = hard_split_days * 86400 groups = [] group = [ordered[0]] group_gaps: list = [] for prev, cur in zip(ordered, ordered[1:]): gap = _epoch(cur.taken_at) - _epoch(prev.taken_at) if gap > hard_cap: split = True elif len(group_gaps) < 2: # bootstrap: accept first 2 gaps split = False else: threshold = max(gap_factor * _median(group_gaps), min_floor_seconds) split = gap > threshold if split: groups.append(group) group = [cur] group_gaps = [] else: group.append(cur) group_gaps.append(gap) groups.append(group) return [_free_cluster(g) for g in groups] def cluster_assets(assets, tags_by_asset, seed_tags, *, gap_factor=6.0, hard_split_days=14, min_floor_seconds=3600) -> list: by_id = {a.immich_id: a for a in assets} used = set() clusters = [] # 1. Seed clusters from existing trip tags (authoritative; never gap-split). for tag in sorted(seed_tags): members = [by_id[aid] for aid in by_id if aid not in used and tag in tags_by_asset.get(aid, [])] if not members: continue members.sort(key=lambda a: a.taken_at) used.update(m.immich_id for m in members) start, end = _span(members) clusters.append(CandidateCluster( member_ids=[m.immich_id for m in members], start_at=start, end_at=end, suggested_name=tag, confidence=SEED_CONFIDENCE, kind_guess="trip", seed_tag=tag)) # 2. Gap-cluster the remaining (free) assets. free = [a for a in assets if a.immich_id not in used] clusters.extend(_gap_cluster(free, gap_factor=gap_factor, hard_split_days=hard_split_days, min_floor_seconds=min_floor_seconds)) clusters.sort(key=lambda c: c.start_at) return clusters # Public aliases (Review revision 2): expose the timestamp/median helpers for # cross-module reuse (e.g. coverage.py) without forcing callers onto the # underscore-prefixed names. The underscore names remain importable. epoch = _epoch median = _median