"""
Implementation of NG911-MSAG comparisons.
"""
from __future__ import annotations

import logging
from collections.abc import Callable, Sequence, MutableMapping
# noinspection PyProtectedMember
from os import PathLike
from pathlib import Path
from typing import Self, Optional, TypedDict, Literal, TypeAlias, Any

import arcpy
import attrs
import pandas as pd
# noinspection PyProtectedMember
# noinspection PyProtectedMember
from attr._make import _CountingAttr
from attrs.converters import pipe, optional
from attrs.validators import ge, in_, optional as opt, instance_of as instof, min_len, max_len, and_
# noinspection PyProtectedMember
from pandas.core.dtypes.base import ExtensionDtype
from pandas.core.dtypes.common import is_extension_array_dtype

from .accessor import RoadRangeCoverage
from .config_dataclasses import NG911Field
from .counties import CountyInfo
from .gdbsession import GPMessenger, NG911SessionWithProduct
from .iterablenamespace import FrozenList
from .misc import Parity, quote, indent, NoneOrNA, na_to_none
from .session import cf, cfc, MSAG_ISSUE_TABLE_NAME, TN_ISSUE_TABLE_NAME


__all__ = ["MSAGComparatorFieldMap", "MSAGFieldMap", "TNFieldMap", "MSAGStreetKeyTuple", "MSAGComparisonIssue", "MSAGIssue", "TNIssue", "NG911MSAGComparator"]

_logger = logging.getLogger(__name__)


@attrs.frozen
class MSAGComparatorFieldMap:
    """
    Base class for spreadsheet field mapping classes.

    A subclass should be written to represent each type of MSAG-related
    spreadsheet. Subclasses should have an ``attrs`` field (created using
    :func:`fm_field`) for each spreadsheet column that corresponds to a
    Next-Generation or Legacy field.

    .. seealso::

        :func:`fm_field`
    """

    @classmethod
    def _get_metadata_entry_dict(cls, key: str) -> dict[str, Any]:
        return {
            attrs_field.name: val
            for attrs_field in attrs.fields(cls)
            if (val := attrs_field.metadata.get(key))
        }

    @classmethod
    def get_dtypes(cls) -> dict[str, ExtensionDtype]:
        return cls._get_metadata_entry_dict("dtype")

    @classmethod
    def get_display_names(cls) -> dict[str, str]:
        return cls._get_metadata_entry_dict("display_name")

    @classmethod
    def get_display_name_list(cls) -> list[str]:
        return [
            attr.metadata["display_name"]
            for attr in attrs.fields(cls)
        ]

    @classmethod
    def get_input_required(cls) -> list[bool]:
        """Returns whether each argument requires an input."""
        return [
            attr.default is attrs.NOTHING
            for attr in attrs.fields(cls)
        ]


def fm_field(dtype: str, display_name: str, default: Any = attrs.NOTHING, **kwargs: Any) -> _CountingAttr:
    """
    Wrapper for :func:`attrs.field` for use with subclasses of
    :class:`MSAGComparatorMap`.

    Naming: :namekey:`[f]ield [m]ap [field]`

    :param dtype: Name of the pandas nullable ``dtype`` appropriate for the
        field being mapped
    :type dtype: str
    :param display_name: Text to show in the geoprocessing tool interface
        describing the field being mapped
    :type display_name: str
    :param default: Default value to pass to :func:`attrs.field`; default
        :data:`attrs.NOTHING`
    :type default: Any
    :param kwargs: Additional keyword arguments to pass to :func:`attrs.field`
    :return: A result from :func:`attrs.field` appropriate for use in a
        subclass of :class:`MSAGComparatorMap`
    :rtype: :class:`attr._make._CountingAttr`
    """
    return attrs.field(
        default=default,
        metadata={"dtype": dtype, "display_name": display_name},
        **kwargs
    )

@attrs.frozen(kw_only=True)
class MSAGFieldMap(MSAGComparatorFieldMap):
    """Field mapping data class for MSAG spreadsheets."""
    add_from: str = fm_field("UInt32", "From-Address")
    add_to: str = fm_field("UInt32", "To-Address")
    parity: str = fm_field("string", "Parity")
    lgcypredir: str = fm_field("string", "Directional Prefix")
    lgcypretyp: Optional[str] = fm_field("string","Street Type Prefix", None)
    lgcystreet: str = fm_field("string", "Street Name")
    lgcytype: Optional[str] = fm_field("string","Street Type Suffix", None)
    lgcysufdir: Optional[str] = fm_field("string","Directional Suffix", None)
    msagcomm: str = fm_field("string", "Community")
    county: str = fm_field("string", "County")
    state: str = fm_field("string", "State")
    esn: str = fm_field("string", "ESN")


@attrs.frozen(kw_only=True)
class TNFieldMap(MSAGComparatorFieldMap):
    addnumber: str = fm_field("UInt32", "Address Number")
    addnumsuf: str = fm_field("string", "Address Suffix")
    lgcypredir: str = fm_field("string", "Directional Prefix")
    lgcypretyp: Optional[str] = fm_field("string", "Street Type Prefix", None)
    lgcystreet: str = fm_field("string", "Street Name")
    lgcytype: Optional[str] = fm_field("string", "Street Type Suffix", None)
    lgcysufdir: Optional[str] = fm_field("string", "Directional Suffix", None)
    msagcomm: str = fm_field("string", "Community")
    county: str = fm_field("string", "County")
    state: str = fm_field("string", "State")
    esn: str = fm_field("string", "ESN")


class _MSAGDataFrameAttributes(TypedDict):
    original_columns: dict[str, str | None]
    """Dictionary of the data frame columns as keys and the column names from
    the spreadsheet (or ``None``, if absent) as their values."""

    source_path: Path
    """Path to the source spreadsheet."""

    source_type: Literal["MSAG", "TN"]
    """
    Type of the source spreadsheet. One of:
    
    - ``MSAG`` for [M]aster [S]treet [A]ddress [G]uide spreadsheet
    - ``TN`` for [T]elephone [N]umber spreadsheet
    """


MSAGStreetKeyTuple: TypeAlias = tuple[str, str, str]
"""Tuple of ``(<street name>, <MSAG community>, <ESN>)``. There should never be
any overlapping address ranges among road centerlines with these values in
common."""


def _validate_and_transform_issue_fields(cls: type, fields: list[attrs.Attribute]) -> list[attrs.Attribute]:
    """Function to be used as the ``field_transformer`` argument to
    :func:`attrs.define`/:func:`attrs.frozen`. This function checks to ensure
    that each field with ``table_field`` in its metadata also has the ``dtype``
    key set to a ``pandas`` extension dtype. It also sets :func:`_na_to_none`
    as the :attr:`attrs.Attribute.converter`. If a converter is already
    present, it is added as the first converter in a
    :func:`attrs.converters.pipe`."""
    for field in fields:
        if "table_field" in field.metadata:
            if "dtype" not in field.metadata:
                raise ValueError(f"{cls.__name__}.{field.name}: 'table_field' key exists in metadata, but 'dtype' key is missing.")
            elif not is_extension_array_dtype(field.metadata["dtype"]):
                raise ValueError(f"{cls.__name__}.{field.name}: dtype '{field.metadata['dtype']}' is not a pandas extension dtype.")
    return [field.evolve(converter=(
        pipe(na_to_none, field.converter)
        if field.converter
        else na_to_none
    )) for field in fields]


@attrs.frozen
class MSAGComparisonIssue:

    source_nguid: str | None = attrs.field(metadata={"table_field": "SrcNGUID", "dtype": "string"})
    """NGUID of the record if it is present in the Next-Generation data,
    otherwise ``None``."""

    source_spreadsheet_row: int | None = attrs.field(
        validator=opt(ge(2)),  # Must be >= 2 if given, because spreadsheets start at row 1 with headers and row 2 is the beginning of the data
        metadata={"table_field": "SrcSpreadsheetRow", "dtype": "Int64"})
    """MSAG or TN row number of the record if it is present in the spreadsheet,
    otherwise ``None``."""

    # @property
    # def metadata_with_values(self) -> dict[tuple[str, FrozenDict[str, Any]], Any]:
    #     """Returns a :class:`~ng911ok.lib.iterablenamespace.FrozenDict` like
    #     so: ``(<field name>, <field attribute dict>): <value>``"""
    #     fields = attrs.fields(self.__class__)  # type: ignore
    #     values_dict = attrs.asdict(self)  # type: ignore
    #     return {(attribute.name, FrozenDict(attribute.metadata)): values_dict[attribute.name] for attribute in fields}

    @property
    def as_dict_by_table_field(self) -> dict[str, Any]:
        """Returns a dictionary that maps the table field name (as given in the
        argument to ``metadata`` in :func:`attrs.field`) of each ``attrs``
        attribute to its value."""
        values = attrs.asdict(self)
        field_md_dict = {f.name: f.metadata for f in attrs.fields(self.__class__)}
        return {
            table_field_name: val
            for attrs_field_name, val
            in values.items()
            if (table_field_name := field_md_dict[attrs_field_name].get("table_field")) is not None
        }
        # return {
        #     table_field_name: val
        #     for (_, metadata), val
        #     in self.metadata_with_values.items()
        #     if (table_field_name := metadata.get("table_field")) is not None
        # }

    def to_series(self) -> pd.Series:
        return pd.Series(self.as_dict_by_table_field, dtype=object)

    @classmethod
    def issues_to_frame(cls, issues: Sequence[Self]) -> pd.DataFrame:
        df = pd.DataFrame(data=[attrs.asdict(issue, filter=lambda a, v: "table_field" in a.metadata and "dtype" in a.metadata) for issue in issues])
        df = df.astype({
            attribute.name: attribute.metadata["dtype"]
            for attribute
            in attrs.fields(cls)
            if attribute.name in df.columns
        }).rename(columns={
            attribute.name: attribute.metadata["table_field"]
            for attribute
            in attrs.fields(cls)
            if attribute.name in df.columns
        })
        return df


@attrs.frozen(field_transformer=_validate_and_transform_issue_fields)
class MSAGIssue(MSAGComparisonIssue):

    side: Optional[Literal["LEFT", "RIGHT"]] = attrs.field(
        validator=in_(["LEFT", "RIGHT", None]),
        metadata={"table_field": "Side", "dtype": "string"})
    add_from: Optional[int] = attrs.field(
        converter=optional(int),
        validator=opt(instof(int)),
        metadata={"table_field": "AddrFrom", "dtype": "Int64"})
    add_to: Optional[int] = attrs.field(
        converter=optional(int),
        validator=opt(instof(int)),
        metadata={"table_field": "AddrTo", "dtype": "Int64"})
    parity: Optional[Parity] = attrs.field(
        metadata={"table_field": "Parity", "dtype": "string"})
    street_name: Optional[str] = attrs.field(
        metadata={"table_field": cf.lgcyfulst.name, "dtype": "string"})
    community: Optional[str] = attrs.field(
        metadata={"table_field": cf.msagcomm.name, "dtype": "string"})
    state: Optional[str] = attrs.field(
        metadata={"table_field": "State", "dtype": "string"})
    esn: Optional[str] = attrs.field(
        metadata={"table_field": "ESN", "dtype": "string"})
    esn_same_lr: Optional[bool] = attrs.field(
        default=None,
        metadata={"table_field": "ESNSameLR", "dtype": "boolean"})
    _message: Optional[str | Callable[[Self], str]] = attrs.field(default=None)
    # This becomes the parameter `message`, NOT `_message`, in __init__(). See
    # https://www.attrs.org/en/23.1.0/init.html#private-attributes-and-aliases

    esn_field: NG911Field | None = attrs.field(
        init=False,
        default=attrs.Factory(lambda self: {"LEFT": cf.esn_l, "RIGHT": cf.esn_r}.get(self.side, None), takes_self=True),  # type: ignore
        metadata={"table_field": "ESNField", "dtype": "string"}
    )
    """Name of the ESN field for the feature/row in question; one of
    :ng911field:`esn`, :ng911field:`esn_l`, or :ng911field:`esn_r`."""

    message: str = attrs.field(
        init=False,
        default=attrs.Factory(lambda self: self._message(self) if callable(self._message) else self._message, takes_self=True),  # type: ignore
        metadata={"table_field": "Message", "dtype": "string"}
    )
    """The message describing the issue."""

    relate_string: str = attrs.field(
        init=False,
        default=attrs.Factory(lambda self: " \u2AF6 ".join([self.street_name, self.community, self.esn]), takes_self=True),  # type: ignore
        metadata={"table_field": "RELATE_STR", "dtype": "string"}
    )

    # @property
    # def address_range(self) -> AddressRange | None:
    #     """Address range for the feature/row in question; derived from
    #     :attr:`add_from`, :attr:`add_to`, and :attr:`parity`."""
    #     return AddressRange(f, t, p, False) if None not in {f := self.add_from, t := self.add_to, p := self.parity} else None

    @classmethod
    def from_msag_row(cls, row: pd.Series, *, side: Literal["LEFT", "RIGHT"] | NoneOrNA = None, message: Callable[[Self], str] | str | NoneOrNA = None) -> Self:
        """Constructor method to be passed to :meth:`pandas.DataFrame.apply`
        with ``axis=1`` to generate instances of this class from a
        :class:`pandas.DataFrame`."""
        row_number: int = row.name  # type: ignore
        msg: Callable[[Self], str] | str | None = None if message is pd.NA else message
        return cls(
            None, row_number,
            side, row["add_from"], row["add_to"], row["parity"],
            row["$name"], row["msagcomm"], row["state"], row["esn"],
            message=message
        )

@attrs.frozen(field_transformer=_validate_and_transform_issue_fields)
class TNIssue(MSAGComparisonIssue):

    full_address: str | NoneOrNA = attrs.field(
        metadata={"table_field": cf.lgcyfuladd.name, "dtype": "string"})
    address_number: str | NoneOrNA = attrs.field(
        converter=str,
        metadata={"table_field": cf.addnumber.name, "dtype": "string"}
    )
    street_name: str | NoneOrNA = attrs.field(
        metadata={"table_field": cf.lgcyfulst.name, "dtype": "string"})
    community: str | NoneOrNA = attrs.field(
        metadata={"table_field": cf.msagcomm.name, "dtype": "string"})
    state: str | NoneOrNA = attrs.field(
        metadata={"table_field": cf.state.name, "dtype": "string"})
    esn: str | NoneOrNA = attrs.field(
        metadata={"table_field": cf.esn.name, "dtype": "string"})
    telephone_number: str | NoneOrNA = attrs.field(
        converter=optional(int),
        validator=opt(and_(min_len(10), max_len(10))),  # If provided, length must be 10

    )
    _message: Optional[str | Callable[[Self], str]] = attrs.field(default=None)
    # This becomes the parameter `message`, NOT `_message`, in __init__(). See
    # https://www.attrs.org/en/23.1.0/init.html#private-attributes-and-aliases

    message: str = attrs.field(
        init=False,
        default=attrs.Factory(lambda self: self._message(self) if callable(self._message) else self._message, takes_self=True),  # type: ignore
        metadata={"table_field": "Message", "dtype": "string"}
    )
    """The message describing the issue."""

    relate_string: str = attrs.field(
        init=False,
        default=attrs.Factory(lambda self: " \u2AF6 ".join([self.street_name, self.community, self.esn]), takes_self=True),  # type: ignore
        metadata={"table_field": "RELATE_STR", "dtype": "string"}
    )

class NG911MSAGComparator(NG911SessionWithProduct, destination_prefix="MSAG_Analysis"):
    """

    """

    def __init__(self,
                 workspace: PathLike[str] | str,
                 respect_submit: bool,
                 use_edit_session: bool = False,
                 messenger: Optional[GPMessenger] = None,
                 destination: Optional[PathLike[str] | str] = None,
                 msag_path: Optional[Path] = None,
                 msag_fields: Optional[MSAGFieldMap] = None,
                 tn_path: Optional[Path] = None,
                 tn_fields: Optional[TNFieldMap] = None,
                 # export: bool = True,
                 **env_kwargs):
        super().__init__(workspace, respect_submit, use_edit_session, messenger, destination, **env_kwargs)

        # destination = Path(destination) if destination is not None else self.gdb_path.parent / self._create_destination_name()
        # if destination.exists():
        #     raise FileExistsError(destination)
        # if not destination.parent.exists():
        #     raise FileNotFoundError(destination)
        # if not destination.parent.is_dir():
        #     raise NotADirectoryError(destination)

        # self._destination: Path = destination
        # """Path to the geodatabase to which analysis output will be written."""

        self.msag_path: Path | None = Path(msag_path) if msag_path is not None else None
        """Path to the MSAG spreadsheet."""

        self.msag_fields: MSAGFieldMap | None = msag_fields
        """Mapping of standard field roles to MSAG column names."""

        self.tn_path: Path | None = Path(tn_path) if tn_path is not None else None
        """Path to the telephone number (TN) spreadsheet."""

        self.tn_fields: TNFieldMap | None = tn_fields
        """Mapping of standard field roles to TN column names."""

        if self.msag_path and not self.msag_path.exists():
            raise FileNotFoundError(f"{self.msag_path} does not exist.")
        if self.tn_path and not self.tn_path.exists():
            raise FileNotFoundError(f"{self.tn_path} does not exist.")

        self._msag_df: pd.DataFrame | None = None
        """Data frame with MSAG spreadsheet data. This is ``None`` until
        :meth:`_load_msag_df` is called."""

        self._tn_df: pd.DataFrame | None = None
        """Data frame with telephone number (TN) spreadsheet data. This is
        ``None`` until :meth:`_load_tn_df` is called."""

        self._issues: list[MSAGIssue] = []
        """List of MSAG issues to report."""

    def __enter__(self) -> Self:
        r = super().__enter__()
        _logger.debug(f"Entered {self.__class__.__name__}.")
        return r

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        super().__exit__(exc_type, exc_val, exc_tb)
        _logger.info(f"Exiting {self.__class__.__name__} with {len(self._issues)} issue(s).")
        # if self.export:
        #     self._export_issue_table()
        _logger.debug(f"Exited {self.__class__.__name__}.")

    def __del__(self) -> None:
        super().__del__()

    # def _create_destination_name(self) -> str:
    #     timestamp_string: str = self._init_timestamp.strftime("%Y%m%d_%H%M%S")
    #     return f"MSAG_Analysis_{timestamp_string}.gdb"

    def _load_spreadsheet_df(self, spreadsheet_type: Literal["MSAG", "TN"]) -> pd.DataFrame:
        """See :attr:`msag_df` and :attr:`tn_df`."""
        # Determine which path and columns to use
        path: Path
        fields: MSAGFieldMap | TNFieldMap
        dtypes: dict[str, ExtensionDtype]
        if spreadsheet_type == "MSAG":
            path = self.msag_path
            fields = self.msag_fields
            dtypes = MSAGFieldMap.get_dtypes()
            if self._msag_df is not None:
                raise RuntimeError(f"MSAG spreadsheet '{path}' has already been loaded.")
        elif spreadsheet_type == "TN":
            path = self.tn_path
            fields = self.tn_fields
            dtypes = TNFieldMap.get_dtypes()
            if self._tn_df is not None:
                raise RuntimeError(f"TN spreadsheet '{path}' has already been loaded.")
        else:
            raise ValueError(f"Invalid spreadsheet type: '{spreadsheet_type}'")

        # Ensure path and fields are set to something
        if path is None:
            raise TypeError(f"No path was provided for spreadsheet type '{spreadsheet_type}'.")
        if fields is None:
            raise TypeError(f"No field mapping was provided for spreadsheet type '{spreadsheet_type}'.")

        # Get column mapping; determine which columns were and weren't mapped
        field_dict: dict[str, str] = attrs.asdict(fields)  # type: ignore
        provided_columns = [c for c in attrs.astuple(fields) if c is not None]  # type: ignore
        omitted_columns: set[str] = set(k for k, v in field_dict.items() if not v)

        # Load in Excel file using provided column names; assign dtypes
        df: pd.DataFrame = pd.read_excel(
            path,
            # usecols=fields.values(),
            usecols=provided_columns,  # type: ignore
            dtype={
                in_col_name: dtypes[role]
                # for role, in_col_name in fields.items()
                for role, in_col_name in field_dict.items()
            }
        )

        # Ensure index corresponds to row number; add 1 because spreadsheets
        # use 1-based indexing instead of 0-based, and add 1 more because the
        # column names are on the first row.
        df.index += 2

        # Create any omitted columns and set to NA to ensure that all columns exist
        for col in omitted_columns:
            df[col] = pd.Series(dtype=dtypes[col])

        # Attach attributes dict to the data frame
        attribute_dict: _MSAGDataFrameAttributes = {
            "source_path": path,
            "source_type": spreadsheet_type,
            "original_columns": field_dict
            # out_col_name: in_col_name
            # for role, in_col_name in fields.items()
            # for out_col_name, in_col_name in field_dict.items()
            # }
        }
        df.attrs = attribute_dict

        # Rename columns from names in the spreadsheet to the names in ``fields``
        df.rename(columns={
            in_col_name: out_col_name
            for out_col_name, in_col_name in attribute_dict["original_columns"].items()
        }, inplace=True)

        # Reorder columns
        df = df[[*field_dict.keys()]]

        # Compute equivalent of LgcyFulSt; automatically ignores NA values
        street_concat_columns: list[str] = ["lgcypredir", "lgcypretyp", "lgcystreet", "lgcytype", "lgcysufdir"]
        df["$name"] = df[street_concat_columns].apply(lambda row: row.str.cat(sep=" "), axis=1)

        if spreadsheet_type == "MSAG":
            # Ensure that the 'parity' column only has appropriate values
            df["parity"] = df["parity"].str.upper().replace({
                "O": "ODD",
                "E": "EVEN",
                "B": "BOTH"
            })
            if not df["parity"].isin(["ODD", "EVEN", "BOTH"]).all():
                unique_values = ", ".join(quote(df["parity"].unique()))
                raise ValueError(f"All values in the MSAG's parity column must be 'ODD', 'EVEN', or 'BOTH' (or 'O', 'E', or 'B'). The following values were present: {unique_values}.")

            range_is_na: pd.Series = df[["add_from", "add_to"]].isna().any(axis=1)
            if range_is_na.any():
                range_na_count = range_is_na.sum()
                self.messenger.addWarningMessage(f"{range_na_count} row(s) in spreadsheet have null values for 'add_from' and/or 'add_to'.")
                data_string: str = df[range_is_na].to_string(
                    max_rows=(max_rows := 20),
                    max_cols=len(df.columns),
                    line_width=254)
                if range_na_count > max_rows:
                    data_string += f"\n[{range_na_count - max_rows} ADDITIONAL ROW(S)]"
                _logger.warning(f"Row(s) with NA in address range:\n{indent(data_string)}")
                na_issues: pd.Series = df[range_is_na].apply(MSAGIssue.from_msag_row, axis=1, message="Null address number(s) in MSAG.")
                self._issues += na_issues.tolist()
                df = df[~range_is_na]

        # If values in the "county" column are strings of digits, assume they
        # represent the three-digit county portion of FIPS codes; replace with
        # county names
        fips_counties: pd.Series = df["county"].str.isdigit()  # Boolean series
        if fips_counties.any():
            df.loc[fips_counties, "county"] = df.loc[fips_counties, "county"].apply(
                lambda val: CountyInfo.get_county(fips3=val).full_name
            )
            _logger.debug(f"Converted {fips_counties.sum()} value(s) in the 'county' column from FIPS codes to county names.")

        _logger.info(f"Loaded {spreadsheet_type} spreadsheet from '{path}'.")
        return df

    def _ensure_destination(self) -> None:
        """Creates the output geodatabase if it doesn't already exist;
        otherwise does nothing. Once this method returns, the output
        geodatabase is sure to exist."""
        if self.destination_path.exists():
            _logger.debug(f"MSAG analysis output geodatabase '{self.destination_path.name}' already exists; no need to create it.")
        else:
            arcpy.management.CreateFileGDB(str(self.destination_path.parent), self.destination_path.name)
            self.messenger.addMessage(f"Created MSAG analysis output geodatabase '{self.destination_path.name}'.")

    def export_msag_issue_table(self) -> Path | None:
        if not self._issues:
            self.messenger.addMessage("No issues to export.")
            return None
        issue_df = MSAGIssue.issues_to_frame(self.msag_issues)
        self._ensure_destination()
        out_table_path = issue_df.ng911.to_table(self.destination_path, MSAG_ISSUE_TABLE_NAME, overwrite=True)
        self.messenger.addMessage(f"Exported {len(issue_df)} issue(s).")
        return out_table_path

    def export_tn_issue_table(self) -> Path | None:
        ...

    # @property
    # def destination(self) -> str:
    #     """Path (as a string) to the analysis output geodatabase."""
    #     return str(self._destination)
    #
    # @property
    # def destination_path(self) -> Path:
    #     """Path to the analysis output geodatabase."""
    #     return self._destination

    @property
    def msag_issue_table_path(self) -> Path:
        """Path to the MSAG issue table (regardless of whether it currently
        exists)."""
        return self._destination / MSAG_ISSUE_TABLE_NAME

    @property
    def tn_issue_table_path(self) -> Path:
        """Path to the TN issue table (regardless of whether it currently
        exists)."""
        return self._destination / TN_ISSUE_TABLE_NAME

    @property
    def msag_df(self) -> pd.DataFrame:
        """
        The first time this property is accessed on an instance, the
        spreadsheet at :attr:`msag_path` is loaded and stored as a
        ``pandas.DataFrame`` (as :attr:`_msag_df`).

        This property returns a copy of that data frame, which has the
        following columns and ``dtype``\ s:

        .. csv-table::
            :header-rows: 1

            Column Name, ``dtype``
            ``add_from``, :class:`UInt32 <pandas.UInt32Dtype>`
            ``add_to``, ``UInt32``
            ``parity``, :class:`string <pandas.StringDtype>` (``ODD``/``EVEN``/``BOTH``)
            ``lgcypredir`` ``string``
            ``lgcypretyp`` ``string``
            ``lgcystreet`` ``string``
            ``lgcytype`` ``string``
            ``lgcysufdir`` ``string``
            ``msagcomm`` ``string``
            ``county`` ``string``
            ``state`` ``string``
            ``esn`` ``string``
            ``$name``, ``string``

        .. note::

            The above ``dtype``\ s are ``pandas`` extension ``dtype``\ s.

        ``$name`` is a column computed in the same way as
        :ng911field:`lgcyfulst`.
        """
        if self._msag_df is None:
            self._msag_df = self._load_spreadsheet_df("MSAG")
        return self._msag_df.copy()

    @property
    def tn_df(self) -> pd.DataFrame:
        """
        The first time this property is accessed on an instance, the
        spreadsheet at :attr:`tn_path` is loaded and stored as a
        ``pd.DataFrame`` (as :attr:`_tn_df`).

        This property returns a copy of that data frame, which has the
        following columns and ``dtype``\ s:

        .. csv-table::
            :header-rows: 1

            Column Name, dtype
            ``addnumber``, :class:`UInt32 <pandas.UInt32Dtype>`
            ``addnumsuf``, :class:`string <pandas.StringDtype>`
            ``predir``, ``string``
            ``street``, ``string``
            ``msagcomm``, ``string``
            ``county``, ``string``
            ``state``, ``string``
            ``esn``, ``string``
        """
        if self._tn_df is None:
            self._tn_df = self._load_spreadsheet_df("TN")
        return self._tn_df.copy()

    @property
    def issues(self) -> FrozenList[MSAGIssue | TNIssue]:
        return FrozenList(self._issues)

    @property
    def msag_issues(self) -> FrozenList[MSAGIssue]:
        return FrozenList([issue for issue in self._issues if isinstance(issue, MSAGIssue)])

    @property
    def tn_issues(self) -> FrozenList[TNIssue]:
        return FrozenList([issue for issue in self._issues if isinstance(issue, TNIssue)])

    def find_missing_msag_entries(self):
        """Finds address ranges that are present in :ng911fc:`road_centerline`
        but not in the MSAG."""
        ...

    def find_missing_address_ranges(self):
        """Finds address ranges that are present in the MSAG but not in
        :ng911fc:`road_centerline`."""
        # msag: pd.DataFrame = self.msag_df
        # roads: pd.DataFrame = self.load_df(cfc.road_centerline)
        # msag["$street_name"] = msag.apply(
        #     lambda row: " ".join(row[row.notna() & ["a", "b"]]) or pd.NA,
        #     axis=1
        # )
        msag_coverages: dict[MSAGStreetKeyTuple, RoadRangeCoverage] = self.compute_msag_coverage()
        road_coverages: dict[MSAGStreetKeyTuple, RoadRangeCoverage] = self.compute_road_centerline_coverage()
        issues: list[MSAGIssue] = []
        for key, coverage in msag_coverages.items():
            road_coverage = road_coverages.get(key, None)
            if road_coverage is None:
                missing_odds = coverage.odd_ranges
                missing_evens = coverage.even_ranges
            else:
                missing_coverage = coverage - road_coverage
                missing_odds = missing_coverage.odd_ranges
                missing_evens = missing_coverage.even_ranges
            for missing_ranges, missing_parity in zip((missing_odds, missing_evens), (Parity.ODD, Parity.EVEN)):
                for missing_from, missing_to in missing_ranges:
                    issues.append(MSAGIssue(
                        None, None, None,
                        missing_from, missing_to, missing_parity,
                        coverage.name, coverage.community, None, coverage.esn,
                        message="Address range present in MSAG but missing from road feature class."
                    ))
        self._issues += issues
        return issues

    def find_missing_tn_entries(self):
        """Finds addresses that are present in :ng911fc:`address_point` but not
        in the TN file."""
        ...

    def find_missing_address_points(self):
        """Finds address that are present in the TN file but not in
        :ng911fc:`address_point`."""
        ...

    def _compute_coverage(self,
                          df: pd.DataFrame,
                          group_keys: MSAGStreetKeyTuple,
                          add_from: str = "add_from",
                          add_to: str = "add_to",
                          parity: str = "parity",
                          *,
                          coverages: Optional[MutableMapping[MSAGStreetKeyTuple, RoadRangeCoverage]] = None
                          ) -> dict[MSAGStreetKeyTuple, RoadRangeCoverage]:
        """
        Computes road ranges coverages from a data frame. Records with null
        values for the from-address and/or to-address are dropped.

        :param df: Data frame from which to compute coverage
        :type df: pd.DataFrame
        :param group_keys: Tuple of column names corresponding to street name,
            MSAG community, and ESN
        :type group_keys: MSAGStreetKeyTuple
        :param add_from: Name of from-address column, default "add_from"
        :type add_from: str
        :param add_to: Name of to-address column, default "add_to"
        :type add_to: str
        :param parity: Name of parity column, default "parity"
        :type parity: str
        :param coverages: Optional mapping to **update in-place**
        :type coverages: Optional[MutableMapping[MSAGStreetKeyTuple, RoadRangeCoverage]]
        :return: Computed coverage dict, which will be the updated *coverages*
            object if it was provided
        :rtype: MutableMapping[MSAGStreetKeyTuple, RoadRangeCoverage]
        """
        coverages = coverages or {}
        for key, group in df.groupby([*group_keys], dropna=False):  # type: MSAGStreetKeyTuple, pd.DataFrame
            name, community, esn = key
            parity_gb = group.groupby(parity)[[add_from, add_to]]
            odds = parity_gb.get_group(Parity.ODD).values if Parity.ODD in parity_gb.groups else None
            evens = parity_gb.get_group(Parity.EVEN).values if Parity.EVEN in parity_gb.groups else None
            boths = parity_gb.get_group(Parity.BOTH).values if Parity.BOTH in parity_gb.groups else None
            coverage = coverages.setdefault(key, RoadRangeCoverage(name, community, esn))
            coverage |= RoadRangeCoverage(name, community, esn, odds, evens, boths)
        return coverages

    def compute_road_centerline_coverage(self) -> dict[MSAGStreetKeyTuple, RoadRangeCoverage]:
        left_ftp_fields = (cf.add_l_from, cf.add_l_to, cf.parity_l)  # ftp = [f]rom-[t]o-[p]arity
        right_ftp_fields = (cf.add_r_from, cf.add_r_to, cf.parity_r)
        left_group_fields = (cf.msagcomm_l, cf.esn_l)
        right_group_fields = (cf.msagcomm_r, cf.esn_r)
        rcl_df: pd.DataFrame = self.load_df(
            cfc.road_centerline,
            [
                cfc.road_centerline.unique_id,
                *left_ftp_fields,
                *left_group_fields,
                *right_ftp_fields,
                *right_group_fields,
                cf.lgcyfulst
            ]
        )
        alf, alt, pl = (f.name for f in left_ftp_fields)  # [a]dd_[l]_[f]rom, [a]dd_[l]_[t]o, [p]arity_[l]
        arf, art, pr = (f.name for f in right_ftp_fields)
        left_group_names: MSAGStreetKeyTuple = tuple(f.name for f in (cf.lgcyfulst, *left_group_fields))  # type: ignore
        right_group_names: MSAGStreetKeyTuple = tuple(f.name for f in [cf.lgcyfulst, *right_group_fields])  # type: ignore

        coverages: dict[MSAGStreetKeyTuple, RoadRangeCoverage] = self._compute_coverage(rcl_df, left_group_names, alf, alt, pl)
        coverages = self._compute_coverage(rcl_df, right_group_names, arf, art, pr, coverages=coverages)

        return coverages

    def compute_msag_coverage(self) -> dict[MSAGStreetKeyTuple, RoadRangeCoverage]:
        return self._compute_coverage(self.msag_df, ("$name", "msagcomm", "esn"))