"""Ordinance jurisdiction info"""
import logging
import importlib.resources
import unicodedata
from warnings import warn
from contextlib import suppress
from functools import cached_property, lru_cache
import numpy as np
import pandas as pd
from compass.exceptions import COMPASSValueError
from compass.warn import COMPASSWarning
logger = logging.getLogger(__name__)
KNOWN_JURISDICTIONS_REGISTRY = {
importlib.resources.files("compass") / "data" / "conus_jurisdictions.csv",
importlib.resources.files("compass") / "data" / "rmp_jurisdictions.csv",
}
_JUR_COLS = [
"Jurisdiction Type",
"State",
"County",
"Subdivision",
"FIPS",
"Website",
]
_JURISDICTION_TYPES_AS_PREFIXES = {
"town",
"township",
"city",
"borough",
"village",
"unorganized territory",
"location",
}
[docs]
class Jurisdiction:
"""Model a geographic jurisdiction used throughout COMPASS
The class normalizes casing for location components and provides
convenience properties for rendering jurisdiction names with
correct prefixes. It is designed to align with ordinance validation
logic that expects consistent casing and phrasing across states,
counties, and municipal subdivisions.
Notes
-----
Instances compare case-insensitively for type and state, while the
county and subdivision name comparisons preserve their stored
casing. Hashing and ``str`` conversions defer to the full display
name generated by :attr:`full_name`.
"""
def __init__(
self,
subdivision_type,
state,
county=None,
subdivision_name=None,
code=None,
website_url=None,
):
"""
Parameters
----------
subdivision_type : str
Type of subdivision that this jurisdiction represents.
Typical values are "state", "county", "town", "city",
"borough", "parish", "township", etc.
state : str
Name of the state containing the jurisdiction.
county : str, optional
Name of the county containing the jurisdiction, if
applicable. If the jurisdiction represents a state, leave
this input unspecified. If the jurisdiction represents a
county or a subdivision within a county, provide the county
name here.
.. IMPORTANT:: Make sure this input is capitalized properly!
By default, ``None``.
subdivision_name : str, optional
Name of the subdivision that the jurisdiction represents, if
applicable. If the jurisdiction represents a state or
county, leave this input unspecified. Otherwise, provide the
jurisdiction name here.
.. IMPORTANT:: Make sure this input is capitalized properly!
By default, ``None``.
code : str, optional
Optional jurisdiction code (typically FIPS or similar).
By default, ``None``.
website_url : str, optional
Optional URL for the jurisdiction's main website.
By default, ``None``.
"""
self.type = subdivision_type.title()
self.state = state.title()
self.county = county
self.subdivision_name = subdivision_name
self.code = None if code is None else str(code)
self.website_url = website_url
[docs]
@cached_property
def full_name(self):
"""str: Comma-separated jurisdiction display name"""
name_parts = [
self.full_subdivision_phrase,
self.full_county_phrase,
self.state,
]
return ", ".join(filter(None, name_parts))
[docs]
@cached_property
def full_name_the_prefixed(self):
"""str: Full location name prefixed with ``the`` as needed"""
if self.type.casefold() == "state":
return f"the state of {self.state}"
if self.type.casefold() in _JURISDICTION_TYPES_AS_PREFIXES:
return f"the {self.full_name}"
return self.full_name
[docs]
@cached_property
def short_name_with_state(self):
"""str: Comma-separated short jurisdiction name"""
if self.subdivision_name:
name_parts = [self.full_subdivision_phrase]
else:
name_parts = [self.full_county_phrase]
name_parts.append(self.state)
return ", ".join(filter(None, name_parts))
[docs]
@cached_property
def short_name_with_state_the_prefixed(self):
"""str: Short jurisdiction name maybe prefixed with ``the``"""
if not self.subdivision_name:
return self.short_name_with_state
if self.type.casefold() in _JURISDICTION_TYPES_AS_PREFIXES:
return f"the {self.short_name_with_state}"
return self.short_name_with_state
[docs]
@cached_property
def full_subdivision_phrase(self):
"""str: Subdivision phrase for the jurisdiction or empty str"""
if not self.subdivision_name:
return ""
if self.type.casefold() in _JURISDICTION_TYPES_AS_PREFIXES:
return f"{self.type} of {self.subdivision_name}"
return f"{self.subdivision_name} {self.type}"
[docs]
@cached_property
def full_subdivision_phrase_the_prefixed(self):
"""str: Subdivision phrase prefixed with ``the`` as needed"""
if self.type.casefold() in _JURISDICTION_TYPES_AS_PREFIXES:
return f"the {self.full_subdivision_phrase}"
return self.full_subdivision_phrase
[docs]
@cached_property
def full_county_phrase(self):
"""str: County phrase for the jurisdiction or empty str"""
if not self.county:
return ""
if not self.subdivision_name:
return f"{self.county} {self.type}"
return f"{self.county} County"
def __repr__(self):
return str(self)
def __str__(self):
return self.full_name
def __eq__(self, other):
if isinstance(other, self.__class__):
return (
self.type.casefold() == other.type.casefold()
and self.state.casefold() == other.state.casefold()
and self.county == other.county
and self.subdivision_name == other.subdivision_name
)
if isinstance(other, str):
return self.full_name.casefold() == other.casefold()
return False
def __hash__(self):
return hash(self.full_name.casefold())
[docs]
def load_all_jurisdiction_info():
"""Load canonical jurisdiction metadata for the continental US
Returns
-------
pandas.DataFrame
Table containing jurisdiction names, FIPS codes, official
websites, and related attributes.
Notes
-----
Missing values are normalized to ``None`` to simplify downstream
serialization. A shallow copy is returned on each call so callers
can safely mutate their local view without affecting the cached
source data.
"""
return _load_all_jurisdiction_info_cached().copy()
@lru_cache(maxsize=1)
def _load_all_jurisdiction_info_cached():
"""Load and cache canonical jurisdiction metadata"""
return pd.concat(
pd.read_csv(fp, dtype=str).replace({np.nan: None})
for fp in KNOWN_JURISDICTIONS_REGISTRY
)
[docs]
def jurisdiction_websites(jurisdiction_info=None):
"""Build a mapping of jurisdiction identifiers to website URLs
Parameters
----------
jurisdiction_info : pandas.DataFrame, optional
DataFrame containing jurisdiction names and websites. If
``None``, this info is loaded using
:func:`load_all_jurisdiction_info`.
By default, ``None``.
Returns
-------
dict
Mapping from jurisdiction FIPS codes to their primary website
URLs.
Notes
-----
The helper uses FIPS codes rather than string names to avoid
collisions between same-named jurisdictions in different states.
"""
if jurisdiction_info is None:
jurisdiction_info = load_all_jurisdiction_info()
return {
row["FIPS"]: row["Website"] for __, row in jurisdiction_info.iterrows()
}
[docs]
def load_jurisdictions_from_subdivision_names(
subdivision_names, state=None, jurisdiction_info=None
):
"""Load known jurisdictions matching subdivision names
Parameters
----------
subdivision_names : str or iterable of str
One or more subdivision names to match against the canonical
jurisdiction data.
state : str, optional
Optional state name used to filter matching jurisdictions.
Matching uses the same normalized comparison applied to the
subdivision names. By default, ``None``.
jurisdiction_info : pandas.DataFrame, optional
DataFrame containing jurisdiction info. If ``None``, this info
is loaded using :func:`load_all_jurisdiction_info`. By default,
``None``.
Returns
-------
pandas.DataFrame
Rows from the canonical jurisdiction data whose subdivision
names match the requested names.
"""
if jurisdiction_info is None:
jurisdiction_info = load_all_jurisdiction_info()
if subdivision_names is None:
subdivision_names = []
elif isinstance(subdivision_names, str):
subdivision_names = [subdivision_names]
normalized_names = {
normalized_name
for name in subdivision_names
if (normalized_name := _normalize_jurisdiction_name(name))
}
if not normalized_names:
return jurisdiction_info.iloc[0:0].copy() # empty df
subdivision_mask = (
jurisdiction_info["Subdivision"]
.map(_normalize_jurisdiction_name)
.isin(normalized_names)
)
if state is not None:
normalized_state = _normalize_jurisdiction_name(state)
subdivision_mask &= (
jurisdiction_info["State"].map(_normalize_jurisdiction_name)
== normalized_state
)
return jurisdiction_info[subdivision_mask].reset_index(drop=True)
[docs]
def load_jurisdictions_from_fp(jurisdiction_fp):
"""Load jurisdiction metadata for entries listed in a CSV file
This loader trims whitespace, deduplicates request rows, and filters
out jurisdictions not present in the canonical data set.
Parameters
----------
jurisdiction_fp : path-like
Path to csv file containing "County" and "State" columns that
define the jurisdictions for which info should be loaded.
Returns
-------
pandas.DataFrame
Jurisdiction information, including FIPS codes and websites,
for every matching entry in the lookup table.
Raises
------
COMPASSValueError
If the input file is missing required columns (``State`` or
``Jurisdiction Type`` when subdivisions are provided).
Notes
-----
Missing jurisdictions trigger warnings with a tabular summary.
"""
jurisdictions = pd.read_csv(jurisdiction_fp, dtype=str).replace(
{np.nan: None}
)
jurisdictions = _validate_jurisdiction_input(jurisdictions)
all_jurisdiction_info = load_all_jurisdiction_info()
merge_cols = ["County", "State"]
if "Subdivision" in jurisdictions:
merge_cols += ["Subdivision", "Jurisdiction Type"]
else:
all_jurisdiction_info = all_jurisdiction_info[
all_jurisdiction_info["Subdivision"].isna()
].reset_index(drop=True)
jurisdictions = ( # remove dupes
jurisdictions.groupby(merge_cols, dropna=False)
.first()
.reset_index()
.drop(columns="Unnamed: 0", errors="ignore")
.replace({np.nan: None})
)
jurisdictions["jur_merge"] = jurisdictions.apply(
_build_merge_col, axis=1, merge_cols=merge_cols
)
all_jurisdiction_info["jur_merge"] = all_jurisdiction_info.apply(
_build_merge_col, axis=1, merge_cols=merge_cols
)
jurisdictions = jurisdictions.merge(
all_jurisdiction_info,
on="jur_merge",
how="left",
suffixes=["_user", ""],
)
jurisdictions = _filter_not_found_jurisdictions(jurisdictions, merge_cols)
return _format_jurisdiction_df_for_output(jurisdictions)
[docs]
def jurisdictions_from_df(jurisdiction_info=None):
"""Convert rows DataFrame into Jurisdiction instances
Parameters
----------
jurisdiction_info : pandas.DataFrame, optional
DataFrame containing jurisdiction info with columns:
``["Jurisdiction Type", "State", "County", "Subdivision",
"FIPS", "Website"]``. If ``None``, this info is loaded using
:func:`load_all_jurisdiction_info`. By default, ``None``.
Yields
------
Jurisdiction
Jurisdiction instance built from each row of the input
DataFrame.
"""
if jurisdiction_info is None:
jurisdiction_info = load_all_jurisdiction_info()
for __, row in jurisdiction_info.iterrows():
jur_type, state, county, sub, fips, website = row[_JUR_COLS]
yield Jurisdiction(
subdivision_type=jur_type,
state=state,
county=county,
subdivision_name=sub,
code=fips,
website_url=website,
)
def _validate_jurisdiction_input(jurisdictions):
"""Throw error if user is missing required columns"""
if "State" not in jurisdictions:
msg = "The jurisdiction input must have at least a 'State' column!"
raise COMPASSValueError(msg)
jurisdictions["State"] = jurisdictions["State"].str.strip()
if "County" not in jurisdictions:
jurisdictions["County"] = None
else:
jurisdictions["County"] = jurisdictions["County"].str.strip()
if "Subdivision" in jurisdictions:
if "Jurisdiction Type" not in jurisdictions:
msg = (
"The jurisdiction input must have a 'Jurisdiction Type' "
"column if a 'Subdivision' column is provided (this helps "
"avoid name clashes for certain subdivisions)!"
)
raise COMPASSValueError(msg)
jurisdictions["Subdivision"] = jurisdictions["Subdivision"].str.strip()
jurisdictions["Jurisdiction Type"] = (
jurisdictions["Jurisdiction Type"].str.casefold().str.strip()
)
return jurisdictions
def _filter_not_found_jurisdictions(df, merge_cols):
"""Filter out jurisdictions with null FIPS codes"""
_warn_about_missing_jurisdictions(df, merge_cols)
return df[~df["FIPS"].isna()].copy()
def _warn_about_missing_jurisdictions(df, merge_cols):
"""Throw warning about jurisdictions that were not in the list"""
not_found_jurisdictions = df[df["FIPS"].isna()]
if len(not_found_jurisdictions):
out_cols = {f"{col}_user": col for col in merge_cols}
not_found_jurisdictions = not_found_jurisdictions[
list(out_cols)
].rename(columns=out_cols)
not_found_jurisdictions_str = not_found_jurisdictions[
merge_cols
# cspell: disable-next-line
].to_markdown(index=False, tablefmt="psql")
msg = (
"The following jurisdictions were not found! Please make sure to "
"use proper spelling and capitalization.\n"
f"{not_found_jurisdictions_str}"
)
warn(msg, COMPASSWarning)
def _format_jurisdiction_df_for_output(df):
"""Format jurisdiction DataFrame for output"""
out_cols = [
"County",
"State",
"Subdivision",
"Jurisdiction Type",
"FIPS",
"Website",
]
df["FIPS"] = df["FIPS"].apply(fips_to_str)
return df[out_cols].replace({np.nan: None}).reset_index(drop=True)
def _build_merge_col(row, merge_cols):
"""Build column to merge jurisdiction DataFrames on"""
return " ".join(str(row[c]).casefold() for c in merge_cols)
def _normalize_jurisdiction_name(name):
"""Normalize jurisdiction names for resilient comparisons"""
if name is None or pd.isna(name):
return ""
normalized_name = unicodedata.normalize("NFKD", str(name).casefold())
return "".join(char for char in normalized_name if char.isalnum())
def fips_to_str(value):
"""[NOT PUBLIC API] Convert FIPS code to string"""
with suppress(ValueError):
value = int(value)
return str(value)
