import json
import logging
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd
from IPython.display import Markdown, display


# Same root resolution as 01_money_lens: nbconvert's cwd is the build dir, so the
# runner passes the root; the cwd search is for running interactively.
def _find_root() -> Path:
    env = os.environ.get("EDMONTON_REPO_ROOT")
    if env:
        return Path(env).resolve()
    here = Path.cwd()
    for p in (here, *here.parents):
        if (p / "src" / "join_and_calculate.py").exists():
            return p
    raise RuntimeError(
        "cannot locate the repo root — set EDMONTON_REPO_ROOT or run from inside the repo"
    )


REPO_ROOT = _find_root()
sys.path.insert(0, str(REPO_ROOT))
import main  # noqa: E402  (import-safe; puts src/ on the path)

# The pipeline's INFO lines are the Money lens notebook's subject, not this
# page's; warnings still surface.
logging.basicConfig(stream=sys.stdout, level=logging.WARNING, format="%(name)s: %(message)s")

CHECKS: list[tuple[bool, str, str]] = []


def check(ok: bool, claim: str, detail: str = "") -> None:
    """Record an invariant. Reported together at the end; never stops the run."""
    CHECKS.append((bool(ok), claim, detail))
    print(f"  [{'PASS' if ok else 'FAIL'}] {claim}{'  — ' + detail if detail else ''}")


def md(text: str) -> None:
    # The notebook renderer reads a pair of bare `$` as TeX math delimiters,
    # which swallows the text between two dollar amounts.
    display(Markdown(text.replace("$", r"\$")))


WEB_DATA = REPO_ROOT / "web" / "data"
features = json.loads((WEB_DATA / "neighbourhood_value_per_acre.geojson").read_text())["features"]
served = pd.DataFrame([f["properties"] for f in features])
status = json.loads((WEB_DATA / "status.json").read_text())
grid = json.loads((WEB_DATA / "value_grid.json").read_text())
grid_df = pd.DataFrame(grid["cells"], columns=grid["columns"])

print(f"served neighbourhoods: {len(served):,}")
print(f"roll year / rate year: {status.get('data_year')} / {status.get('rate_year')}")
print(f"served data generated: {status.get('generated')}")
print(f"executed:              {datetime.now(timezone.utc):%Y-%m-%d %H:%M UTC}")

served neighbourhoods: 406
roll year / rate year: 2026 / 2026
served data generated: 2026-10-12
executed:              2026-10-12 17:25 UTC

from apply_tax_rates import apply_tax_rates  # noqa: E402
from load_assessment import load_assessment  # noqa: E402

YEAR = main.ASSESSMENT_YEAR
rates = json.loads(main.MILL_RATES_JSON.read_text())["rates"][str(YEAR)]
assessment = apply_tax_rates(load_assessment(str(main.ASSESSMENT_CSV)), main.MILL_RATES_JSON, YEAR)
levy_total = assessment["levy"].sum()
res_rate = rates["Residential"]["municipal"]
nonres_rate = rates["Non Residential"]["municipal"]

md(f"""
For **{YEAR}** the rates are Residential **{res_rate}**, Other Residential
**{rates['Other Residential']['municipal']}** and Non Residential
**{nonres_rate}** per $1,000. Applied to **{len(assessment):,}** accounts, that
gives a citywide municipal levy of **${levy_total / 1e9:,.2f}B**.

The gap between the value view and the revenue view is exactly Edmonton's
class-differential rates: non-residential property is taxed
**{nonres_rate / res_rate:.1f}×** residential per dollar of value.
(`docs/SPEC_revenue.md`)
""")

apply_tax_rates: 80 propert(ies) with class % not summing to 100 — billed as-stated, not normalized

from export_value_grid import (  # noqa: E402
    KNOWN_BOUND_OUTLIERS, MULTI_UNIT_MIN_LOT_M2, SHARE_MAX_M2, SQ_M_PER_ACRE,
    _point_lot_stats,
)
from join_and_calculate import LOW_PARCEL_FRAC  # noqa: E402
from load_boundaries import load_boundaries  # noqa: E402
from load_property_info import load_property_info  # noqa: E402

boundaries = load_boundaries(str(main.BOUNDARIES_GEOJSON))
ground_acres = boundaries.set_index("neighbourhood_name")["area_acres"]
served["ground_acres"] = served["neighbourhood_name"].map(ground_acres)

check(
    served["ground_acres"].notna().all(),
    "every served neighbourhood has a boundary area",
    f"{served['ground_acres'].isna().sum()} without",
)

# A hood's lot-acre figure is shown only at or above the parcel-coverage floor.
suppressed = served["value_per_lot_acre"].isna()
below_floor = ~(served["parcel_frac"] >= LOW_PARCEL_FRAC)
check(
    not (~suppressed & below_floor).any(),
    "no lot-acre figure is served below the parcel-coverage floor",
    f"{int((~suppressed & below_floor).sum())} served below {LOW_PARCEL_FRAC:.0%}",
)

md(f"""
The map covers **{len(served):,}** neighbourhoods and
**{served['ground_acres'].sum():,.0f}** ground acres. The lot-acre floor is
**{LOW_PARCEL_FRAC:.0%}**: this week **{int(below_floor.sum())}**
neighbourhoods have titled lots covering less than that share of their boundary,
and are greyed out in lot-acre mode.
""")

  [PASS] every served neighbourhood has a boundary area  — 0 without
  [PASS] no lot-acre figure is served below the parcel-coverage floor  — 0 served below 15%

biggest = assessment.nlargest(1, "assessed_value").iloc[0]
hood = biggest["neighbourhood_name"]
hood_value = assessment.loc[assessment["neighbourhood_name"] == hood, "assessed_value"].sum()

# The account's own cell: grid lon/lat are cell centres, so the nearest centre
# (with longitude shrunk by cos(latitude)) is the cell it was binned into.
coslat = np.cos(np.radians(biggest["latitude"]))
dist = np.hypot((grid_df["lon"] - biggest["longitude"]) * coslat, grid_df["lat"] - biggest["latitude"])
its_cell = grid_df.loc[dist.idxmin()]
ground_rank = int((grid_df["revenue_per_acre"] > its_cell["revenue_per_acre"]).sum()) + 1
lot_rank = int((grid_df["revenue_per_lot_acre"] > its_cell["revenue_per_lot_acre"]).sum()) + 1
top_lot = grid_df.nlargest(1, "revenue_per_lot_acre").iloc[0]

md(f"""
This week the largest account is **${biggest['assessed_value'] / 1e9:,.2f}B** in
**{hood.title()}**, where it is **{biggest['assessed_value'] / hood_value:.0%}**
of the neighbourhood's assessed value.

On the grid, its cell carries **${its_cell['revenue_per_acre']:,.0f}** of levy
per ground acre — number **{ground_rank:,}** of {len(grid_df):,} cells. Per acre
of lot it carries **${its_cell['revenue_per_lot_acre']:,.0f}** and ranks
**{lot_rank:,}**, because its lot is large; the top lot-acre cell carries
**${top_lot['revenue_per_lot_acre']:,.0f}**, about
**{top_lot['revenue_per_lot_acre'] / its_cell['revenue_per_lot_acre']:,.0f}×**
more. The two rankings disagree by design — the toggle exists so both readings
can be seen.
""")
grid_df.nlargest(5, "revenue_per_acre")[["lat", "lon", "revenue_per_acre", "revenue_per_lot_acre"]]

lots = assessment.merge(load_property_info(main.PROPERTY_INFO_CSV), on="account_number", how="left")
pts = lots.loc[lots["latitude"].notna() & lots["longitude"].notna()]
per_point = _point_lot_stats(pts[["latitude", "longitude", "lot_size"]])

rows = pts.merge(per_point[["latitude", "longitude", "eligible"]], on=["latitude", "longitude"], how="left")
# A point the classifier never saw is a silent drop; the two sums below would
# still add to the roll, so this is the check, not their total.
check(
    rows["eligible"].notna().all(),
    "every located account's point is classified as in the lot-acre view or excluded",
    f"{int(rows['eligible'].isna().sum())} unclassified",
)
eligible = rows["eligible"].fillna(False).astype(bool)
inelig_value = rows.loc[~eligible, "assessed_value"].sum()
elig_value = rows.loc[eligible, "assessed_value"].sum()
roll_value = pts["assessed_value"].sum()

multi = per_point["n"] > 1
n_inelig = int((~per_point["eligible"]).sum())
elig_acres = per_point.loc[per_point["eligible"], "lot_m2"].sum() / SQ_M_PER_ACRE

md(f"""
The share threshold is **{SHARE_MAX_M2:,.0f} m²**, and a multi-unit point
totalling under **{MULTI_UNIT_MIN_LOT_M2:,.0f} m²** is treated as share-coded
rather than an area. This week the roll has **{len(per_point):,}** distinct
property points, **{int(multi.sum()):,}** of them multi-unit. **{n_inelig:,}**
points are excluded from the lot-acre view, carrying
**${inelig_value / 1e9:,.2f}B** — **{inelig_value / roll_value:.2%}** of located
assessed value.

The eligible lots total **{elig_acres:,.0f}** acres, which is
**{elig_acres / served['ground_acres'].sum():.0%}** of the ground acres above;
the rest is roads, parks, rights-of-way and land with no usable lot record.

A pipeline check asserts each neighbourhood's deduplicated lot acres fit inside
its boundary. Known, documented exceptions:
**{', '.join(o.title() for o in KNOWN_BOUND_OUTLIERS) or 'none'}**.
""")

export_value_grid: 2 multi-record point(s) carry under 10 m² of total lot area — share-coded lot_size, not areas; excluded from the lot-acre metric (examples: [{'latitude': 53.45829395304915, 'longitude': -113.5825300675948, 'n': 6, 'lot_m2': 0.093}, {'latitude': 53.55078641357152, 'longitude': -113.53289287458574, 'n': 3, 'lot_m2': 1.0}])

  [PASS] every located account's point is classified as in the lot-acre view or excluded  — 0 unclassified

from load_zoning import SET_ASIDE_THRESHOLD  # noqa: E402

set_aside = served["is_set_aside"].fillna(False).astype(bool)
check(
    set_aside.equals(served["set_aside_frac"].fillna(0) >= SET_ASIDE_THRESHOLD),
    "the grey flag is exactly the set-aside share at or above the threshold",
    f"{int(set_aside.sum())} flagged",
)
md(f"""
The threshold is **{SET_ASIDE_THRESHOLD:.0%}** of the land. This week
**{int(set_aside.sum())}** of **{len(served):,}** neighbourhoods are grey.
""")

  [PASS] the grey flag is exactly the set-aside share at or above the threshold  — 48 flagged

md(f"""
The cells are **{grid['cell_m']:,.0f} m** on a side; **{len(grid_df):,}** of
them hold at least one account this week.
""")

from load_roads import METRIC_GROUPS  # noqa: E402

from join_and_calculate import load_unit_costs  # noqa: E402

unit_costs = load_unit_costs(main.UNIT_COSTS_JSON)
n_stations = len(json.loads((WEB_DATA / "fire_stations.json").read_text())["stations"])


def citywide(col: str) -> float:
    return (served[col] * served["ground_acres"]).sum()


road_km = citywide("road_m_per_acre") / 1000
check(road_km > 0, "the served road metric covers a positive length", f"{road_km:,.0f} km")

md(f"""
This week, in figures:

| layer | citywide |
|---|---|
| road centreline in the metric ({' + '.join(METRIC_GROUPS)}) | **{road_km:,.0f} km** |
| lifecycle road rate | **${unit_costs['road_dollars_per_m']:,.2f}** per road metre per year |
| operating road rate | **${unit_costs['road_ops_dollars_per_m']:,.2f}** per road metre per year |
| fire events (in development) | **{citywide('fire_events_per_acre'):,.0f}** per year, window **{min(main.FIRE_YEARS)}–{max(main.FIRE_YEARS)}**, **{n_stations}** stations |
| stormwater, modelled (in development) | **${citywide('storm_charge_per_acre') / 1e6:,.1f}M** per year |
| water + sanitary, modelled (in development) | **${citywide('water_charge_per_acre') / 1e6:,.1f}M** per year |
""")

  [PASS] the served road metric covers a positive length  — 3,656 km

md(f"""
This week's vintages, from the site's own manifest: roll **{status.get('data_year')}**,
rates **{status.get('rate_year')}**, zoning **{status.get('zoning_year')}**;
data last checked **{status.get('last_checked')}**.
""")

failed = [c for c in CHECKS if not c[0]]
print(f"{len(CHECKS) - len(failed)}/{len(CHECKS)} invariants held\n")
for ok, claim, detail in CHECKS:
    print(f"  [{'PASS' if ok else 'FAIL'}] {claim}{'  — ' + detail if detail else ''}")

if failed:
    raise AssertionError(
        f"{len(failed)} invariant(s) failed:\n"
        + "\n".join(f"  - {claim} {detail}" for _, claim, detail in failed)
    )
print("\nAll invariants held.")

5/5 invariants held

  [PASS] every served neighbourhood has a boundary area  — 0 without
  [PASS] no lot-acre figure is served below the parcel-coverage floor  — 0 served below 15%
  [PASS] every located account's point is classified as in the lot-acre view or excluded  — 0 unclassified
  [PASS] the grey flag is exactly the set-aside share at or above the threshold  — 48 flagged
  [PASS] the served road metric covers a positive length  — 3,656 km

All invariants held.

	lat	lon	revenue_per_acre	revenue_per_lot_acre
4793	53.521560	-113.623821	13117253	302525.0
18763	53.546121	-113.497700	6335745	15443990.0
15378	53.520372	-113.524264	5890539	666212.0
18762	53.545223	-113.497731	5557683	1464577.0
14817	53.517731	-113.528883	4467964	959056.0

Input	Edmonton Open Data ID	Notes
Property assessments	`q7d6-ambg`	the current roll, live weekly feed
Property information (lot sizes)	`dkk9-cj3x`	joins to assessments by account
Neighbourhood boundaries	`65fr-66s6`
Tax rates	`pwis-wc4c`	municipal mill rates by class
Zoning Bylaw geometry	`fixa-tstc`	land-use categories, set-aside layer
Road network	`9j8t-zm52`	centreline segments
Fire response events / stations	`7hsn-idqi` / `b4y7-zhnz`
Utility tariffs	EPCOR bylaw schedules	year-pinned JSON in `data/`

Methods & definitions¶

1. The core metric: municipal revenue (or assessed value) per acre¶

2. The two hard cases, worked openly¶

3. Grey neighbourhoods: the set-aside layer¶

4. The Glass view (100 m grid)¶

5. The cost side (Services view)¶

6. Display honesty rules¶

7. Data sources¶

8. Known limitations¶

What this page cites rather than recomputes¶

layer	citywide
road centreline in the metric (collector + local)	3,656 km
lifecycle road rate	$50.00 per road metre per year
operating road rate	$9.32 per road metre per year
fire events (in development)	88,047 per year, window 2023–2025, 31 stations
stormwater, modelled (in development)	$256.8M per year
water + sanitary, modelled (in development)	$588.3M per year