Enable UK local authority simulations by duplicating UK constituency breakdown for local areas

.vscode/settings.json

@@ -0,0 +1,5 @@
+{
+  "python-envs.defaultEnvManager": "ms-python.python:conda",
+  "python-envs.defaultPackageManager": "ms-python.python:conda",
+  "python-envs.pythonProjects": []
+}

changelog_entry.yaml

@@ -0,0 +1,4 @@
+- bump: minor
+  changes:
+    added:
+    - Local authority impacts

policyengine/outputs/macro/comparison/calculate_economy_comparison.py

@@ -775,6 +775,74 @@ def uk_constituency_breakdown(
     return UKConstituencyBreakdownWithValues(**output)
 
 
+class UKLocalAreaBreakdownByLocalArea(BaseModel):
+    average_household_income_change: float
+    relative_household_income_change: float
+    x: int
+    y: int
+
+
+class UKLocalAreaBreakdownWithValues(BaseModel):
+    by_local_area: dict[str, UKLocalAreaBreakdownByLocalArea]
+
+
+UKLocalAreaBreakdown = UKLocalAreaBreakdownWithValues | None
+
+
+def uk_local_area_breakdown(
+    baseline: SingleEconomy, reform: SingleEconomy, country_id: str
+) -> UKLocalAreaBreakdown:
+    if country_id != "uk":
+        return None
+
+    output = {
+        "by_local_area": {},
+    }
+    # Note: Country-level aggregation (outcomes_by_region) removed for local areas
+    baseline_hnet = baseline.household_net_income
+    reform_hnet = reform.household_net_income
+
+    local_area_weights_local_path = download(
+        gcs_bucket="policyengine-uk-data-private",
+        gcs_key="local_authority_weights.h5",
+    )
+    with h5py.File(local_area_weights_local_path, "r") as f:
+        weights = f["2025"][
+            ...
+        ]  # {2025: array(n_local_areas, n_households) where cell i, j is the weight of household record j in local area i}
+
+    local_area_names_local_path = download(
+        gcs_bucket="policyengine-uk-data-private",
+        gcs_key="local_authorities_2021.csv",
+    )
+    local_area_names = pd.read_csv(
+        local_area_names_local_path
+    )  # columns code (local area code), name (local area name), x, y (geographic position)
+
+    for i in range(len(local_area_names)):
+        name: str = local_area_names.iloc[i]["name"]
+        code: str = local_area_names.iloc[i]["code"]
+        weight: np.ndarray = weights[i]
+        baseline_income = MicroSeries(baseline_hnet, weights=weight)
+        reform_income = MicroSeries(reform_hnet, weights=weight)
+        average_household_income_change: float = (
+            reform_income.sum() - baseline_income.sum()
+        ) / baseline_income.count()
+        percent_household_income_change: float = (
+            reform_income.sum() / baseline_income.sum() - 1
+        )
+        output["by_local_area"][name] = {
+            "average_household_income_change": average_household_income_change,
+            "relative_household_income_change": percent_household_income_change,
+            "x": int(local_area_names.iloc[i]["x"]),  # Geographic positions
+            "y": int(local_area_names.iloc[i]["y"]),
+        }
+
+        # Note: Country-level aggregation and bucketing logic removed for local areas
+
+    return UKLocalAreaBreakdownWithValues(**output)
+
+
 class CliffImpactInSimulation(BaseModel):
     cliff_gap: float
     cliff_share: float
@@ -802,6 +870,7 @@ class EconomyComparison(BaseModel):
     intra_wealth_decile: IntraWealthDecileImpact
     labor_supply_response: LaborSupplyResponse
     constituency_impact: UKConstituencyBreakdown
+    local_authority_impact: UKLocalAreaBreakdown
     cliff_impact: CliffImpact | None
 
 
@@ -832,6 +901,9 @@ def calculate_economy_comparison(
     constituency_impact_data: UKConstituencyBreakdown = (
         uk_constituency_breakdown(baseline, reform, country_id)
     )
+    local_authority_impact_data: UKLocalAreaBreakdown = (
+        uk_local_area_breakdown(baseline, reform, country_id)
+    )
     wealth_decile_impact_data = wealth_decile_impact(
         baseline, reform, country_id
     )
@@ -870,5 +942,6 @@ def calculate_economy_comparison(
         intra_wealth_decile=intra_wealth_decile_impact_data,
         labor_supply_response=labor_supply_response_data,
         constituency_impact=constituency_impact_data,
+        local_authority_impact=local_authority_impact_data,
         cliff_impact=cliff_impact,
     )

policyengine/simulation.py

@@ -356,7 +356,7 @@ def _apply_region_to_simulation(
                 )
 
                 with h5py.File(weights_local_path, "r") as f:
-                    weights = f[str(self.time_period)][...]
+                    weights = f[str(time_period)][...]
 
                 simulation.set_input(
                     "household_weight",

tests/country/test_uk.py

@@ -52,3 +52,105 @@ def test_uk_macro_bad_data_version_fails():
             },
             data_version="a",
         )
+
+
+def test_uk_macro_comparison_has_local_authority_impact():
+    """Test that UK macro comparison includes local authority breakdown."""
+    from policyengine import Simulation
+
+    sim = Simulation(
+        scope="macro",
+        country="uk",
+        reform={
+            "gov.hmrc.income_tax.allowances.personal_allowance.amount": 15_000,
+        },
+    )
+
+    result = sim.calculate_economy_comparison()
+
+    # Check local_authority_impact is present and not None for UK
+    assert hasattr(result, "local_authority_impact")
+    assert result.local_authority_impact is not None
+
+    # Check structure
+    la_impact = result.local_authority_impact
+    assert hasattr(la_impact, "by_local_area")
+    assert len(la_impact.by_local_area) > 0
+
+    # Check each local authority has required fields
+    for name, data in la_impact.by_local_area.items():
+        assert hasattr(data, "average_household_income_change")
+        assert hasattr(data, "relative_household_income_change")
+        assert hasattr(data, "x")
+        assert hasattr(data, "y")
+        assert isinstance(data.average_household_income_change, float)
+        assert isinstance(data.relative_household_income_change, float)
+        assert isinstance(data.x, int)
+        assert isinstance(data.y, int)
+
+
+def test_uk_macro_comparison_constituency_and_local_authority_both_present():
+    """Test that both constituency and local authority impacts are present."""
+    from policyengine import Simulation
+
+    sim = Simulation(
+        scope="macro",
+        country="uk",
+        reform={
+            "gov.hmrc.income_tax.allowances.personal_allowance.amount": 15_000,
+        },
+    )
+
+    result = sim.calculate_economy_comparison()
+
+    # Both should be present for UK
+    assert result.constituency_impact is not None
+    assert result.local_authority_impact is not None
+
+    # Constituencies should have outcomes_by_region, local authorities should not
+    assert hasattr(result.constituency_impact, "outcomes_by_region")
+    assert not hasattr(result.local_authority_impact, "outcomes_by_region")
+
+
+def test_uk_simulation_local_authority_region_filter_by_code():
+    """Test that simulation can filter by local authority code."""
+    from policyengine import Simulation
+
+    # Use a known local authority code (Westminster)
+    sim = Simulation(
+        scope="macro",
+        country="uk",
+        region="local_authority/E09000033",
+    )
+
+    # Should not raise an error
+    result = sim.calculate_single_economy()
+    assert result is not None
+
+
+def test_uk_simulation_local_authority_region_filter_by_name():
+    """Test that simulation can filter by local authority name."""
+    from policyengine import Simulation
+
+    # Use a known local authority name
+    sim = Simulation(
+        scope="macro",
+        country="uk",
+        region="local_authority/Westminster",
+    )
+
+    # Should not raise an error
+    result = sim.calculate_single_economy()
+    assert result is not None
+
+
+def test_uk_simulation_invalid_local_authority_raises_error():
+    """Test that invalid local authority raises ValueError."""
+    from policyengine import Simulation
+
+    with pytest.raises(ValueError, match="Local authority .* not found"):
+        Simulation(
+            scope="macro",
+            country="uk",
+            region="local_authority/InvalidLocalAuthority123",
+        )

tests/utils/test_maps.py

@@ -0,0 +1,77 @@
+import pytest
+
+
+def test_get_location_options_table_parliamentary_constituencies():
+    """Test loading parliamentary constituencies location table."""
+    from policyengine.utils.maps import get_location_options_table
+
+    df = get_location_options_table("parliamentary_constituencies")
+
+    assert len(df) > 0
+    assert "name" in df.columns
+    assert "code" in df.columns
+    assert "x" in df.columns
+    assert "y" in df.columns
+
+
+def test_get_location_options_table_local_authorities():
+    """Test loading local authorities location table."""
+    from policyengine.utils.maps import get_location_options_table
+
+    df = get_location_options_table("local_authorities")
+
+    assert len(df) > 0
+    assert "name" in df.columns
+    assert "code" in df.columns
+    assert "x" in df.columns
+    assert "y" in df.columns
+
+
+def test_plot_hex_map_local_authorities():
+    """Test plotting hex map for local authorities."""
+    from policyengine.utils.maps import (
+        get_location_options_table,
+        plot_hex_map,
+    )
+
+    # Get local authority names
+    df = get_location_options_table("local_authorities")
+
+    # Create dummy values for each local authority
+    value_by_area_name = {name: i * 0.01 for i, name in enumerate(df["name"])}
+
+    fig = plot_hex_map(value_by_area_name, "local_authorities")
+
+    # Check that a figure was returned
+    assert fig is not None
+    assert hasattr(fig, "data")
+    assert len(fig.data) > 0
+
+
+def test_plot_hex_map_parliamentary_constituencies():
+    """Test plotting hex map for parliamentary constituencies."""
+    from policyengine.utils.maps import (
+        get_location_options_table,
+        plot_hex_map,
+    )
+
+    # Get constituency names
+    df = get_location_options_table("parliamentary_constituencies")
+
+    # Create dummy values for each constituency
+    value_by_area_name = {name: i * 0.01 for i, name in enumerate(df["name"])}
+
+    fig = plot_hex_map(value_by_area_name, "parliamentary_constituencies")
+
+    # Check that a figure was returned
+    assert fig is not None
+    assert hasattr(fig, "data")
+    assert len(fig.data) > 0
+
+
+def test_plot_hex_map_invalid_location_type():
+    """Test that invalid location type raises ValueError."""
+    from policyengine.utils.maps import plot_hex_map
+
+    with pytest.raises(ValueError, match="Invalid location_type"):
+        plot_hex_map({"area": 1.0}, "invalid_type")