Detecting overlay changes that leave the base code untouched

def missed_overlay_amendments(store, since):
    """Compare overlay membership between consecutive versions of each parcel and
    count the transitions the change feed never reported."""
    missed = []
    for parcel_id, versions in store.iter_version_pairs(since=since):
        for prev, cur in versions:
            if prev.base_code == cur.base_code and \
                    set(prev.overlay_codes) != set(cur.overlay_codes):
                reported = store.change_event_exists(parcel_id, cur.valid_from)
                if not reported:
                    missed.append({
                        "parcel_id": parcel_id,
                        "effective": cur.valid_from,
                        "added": sorted(set(cur.overlay_codes) - set(prev.overlay_codes)),
                        "removed": sorted(set(prev.overlay_codes) - set(cur.overlay_codes)),
                    })
    return missed

import hashlib
import json


def change_digest(version) -> str:
    """The canonical identity of a parcel's regulatory state.

    Overlays are SORTED, because a portal that returns them in a different order on
    two consecutive runs would otherwise produce a change event every run. They are
    a set, not a sequence — order carries no meaning here.
    """
    payload = {
        "base": version.base_code,
        "overlays": sorted(version.overlay_codes),
        "effective": version.valid_from.isoformat(),
        "geometry": version.canonical_geometry_digest,
    }
    return hashlib.sha256(
        json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
    ).hexdigest()

OVERLAY_ONLY = "overlay_only"


def classify(prev, cur) -> str:
    if prev is None:
        return "new_parcel"
    base_changed = prev.base_code != cur.base_code
    overlays_changed = set(prev.overlay_codes) != set(cur.overlay_codes)
    geom_changed = prev.canonical_geometry_digest != cur.canonical_geometry_digest

    if geom_changed and not (base_changed or overlays_changed):
        return "boundary_adjustment"
    if base_changed and overlays_changed:
        return "rezone_with_overlay_change"
    if base_changed:
        return "rezone"
    if overlays_changed:
        return OVERLAY_ONLY          # the class that used to be invisible
    return "no_change"

def describe_overlay_change(prev, cur, resolver) -> str:
    before = resolver.resolve(prev)
    after = resolver.resolve(cur)

    added = sorted(set(cur.overlay_codes) - set(prev.overlay_codes))
    removed = sorted(set(prev.overlay_codes) - set(cur.overlay_codes))

    lost_uses = sorted(before.uses - after.uses)
    gained_uses = sorted(after.uses - before.uses)
    numeric = {k: (before.numeric.get(k), v)
               for k, v in after.numeric.items() if before.numeric.get(k) != v}

    parts = []
    if added:
        parts.append("overlay added: " + ", ".join(added))
    if removed:
        parts.append("overlay removed: " + ", ".join(removed))
    if lost_uses:
        parts.append(f"{len(lost_uses)} use(s) no longer permitted: "
                     + ", ".join(lost_uses[:3]) + ("…" if len(lost_uses) > 3 else ""))
    if gained_uses:
        parts.append(f"{len(gained_uses)} use(s) newly permitted")
    for k, (was, now) in numeric.items():
        parts.append(f"{k}: {was} → {now}")
    return "; ".join(parts) or "overlay membership changed with no effect on standards"

def group_by_amendment(changes):
    """900 parcels sharing an effective date and an overlay delta is ONE ordinance.
    Grouping before notification is what keeps the alert channel usable."""
    from collections import defaultdict
    groups = defaultdict(list)
    for ch in changes:
        if ch.change_class != OVERLAY_ONLY:
            continue
        key = (ch.effective_date, tuple(ch.overlays_added), tuple(ch.overlays_removed))
        groups[key].append(ch.parcel_id)
    return [{"effective": k[0], "added": list(k[1]), "removed": list(k[2]),
             "parcels": v, "parcel_count": len(v)} for k, v in groups.items()]

def test_overlay_only_change_is_detected():
    prev = version(base="R-1", overlays=["HO"])
    cur = version(base="R-1", overlays=["HO", "WP"])
    assert classify(prev, cur) == OVERLAY_ONLY
    assert change_digest(prev) != change_digest(cur)


def test_overlay_order_is_not_a_change():
    a = version(base="R-1", overlays=["HO", "WP"])
    b = version(base="R-1", overlays=["WP", "HO"])
    assert classify(a, b) == "no_change"
    assert change_digest(a) == change_digest(b)      # sorted before hashing


def test_description_names_the_effect_not_the_token(resolver):
    prev = version(base="R-1", overlays=[])
    cur = version(base="R-1", overlays=["WP"])
    text = describe_overlay_change(prev, cur, resolver)
    assert "WP" in text
    assert "no longer permitted" in text             # the part a subscriber can act on


def test_one_ordinance_is_one_notification():
    changes = [overlay_change(parcel=f"p{i}", effective="2026-06-01", added=["WP"])
               for i in range(900)]
    groups = group_by_amendment(changes)
    assert len(groups) == 1 and groups[0]["parcel_count"] == 900