"""
Implementation of NG911-MSAG comparisons.
"""
import logging
from collections.abc import Callable, Sequence, MutableMapping, Collection
from os import PathLike
from pathlib import Path
from typing import Self, Optional, TypedDict, Literal, TypeAlias, SupportsInt, Any

import arcpy
import attrs
import numpy as np
import pandas as pd
# noinspection PyProtectedMember
from arcpy._mp import Layer as APRXLayer  # For type hinting
from attr.converters import pipe
from attrs.validators import ge, in_, optional as opt, instance_of as instof
from pandas._libs.missing import NAType
from pandas.core.dtypes.base import ExtensionDtype

from .accessor import AddressRange, RoadRangeCoverage
from .config_dataclasses import NG911Field
from .counties import CountyInfo
from .gdbsession import NG911Session, GPMessenger
from .iterablenamespace import FrozenList, FrozenDict
from .misc import Parity, quote, indent
from .session import cf, cfc, MSAG_ISSUE_TABLE_NAME, TN_ISSUE_TABLE_NAME

_logger = logging.getLogger(__name__)


@attrs.frozen
class MSAGComparatorFieldMap:
    @classmethod
    def get_dtypes(cls) -> dict[str, ExtensionDtype]:
        return {
            attr.name: attr.metadata["dtype"]
            for attr in attrs.fields(cls)
        }

    @classmethod
    def get_display_names(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)
        ]


@attrs.frozen(kw_only=True)
class MSAGFieldMap(MSAGComparatorFieldMap):
    add_from: str = attrs.field(metadata={"dtype": "UInt32", "display_name": "From-Address"})
    add_to: str = attrs.field(metadata={"dtype": "UInt32", "display_name": "To-Address"})
    parity: str = attrs.field(metadata={"dtype": "string", "display_name": "Parity"})
    lgcypredir: str = attrs.field(metadata={"dtype": "string", "display_name": "Directional Prefix"})
    lgcypretyp: Optional[str] = attrs.field(default=None, metadata={"dtype": "string", "display_name": "Street Type Prefix"})
    lgcystreet: str = attrs.field(metadata={"dtype": "string", "display_name": "Street Name"})
    lgcytype: Optional[str] = attrs.field(default=None, metadata={"dtype": "string", "display_name": "Street Type Suffix"})
    lgcysufdir: Optional[str] = attrs.field(default=None, metadata={"dtype": "string", "display_name": "Directional Suffix"})
    msagcomm: str = attrs.field(metadata={"dtype": "string", "display_name": "Community"})
    county: str = attrs.field(metadata={"dtype": "string", "display_name": "County"})
    state: str = attrs.field(metadata={"dtype": "string", "display_name": "State"})
    esn: str = attrs.field(metadata={"dtype": "string", "display_name": "ESN"})


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

    @classmethod
    def get_dtypes(cls) -> dict[str, ExtensionDtype]:
        return {
            attr.name: attr.metadata["dtype"]
            for attr in attrs.fields(TNFieldMap)  # type: ignore
        }


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."""


NoneOrNA: TypeAlias = None | NAType
def _na_to_none(val):
    return None if val is pd.NA else val
def _int_or_none(val) -> int | None:
    return int(val) if isinstance(val, SupportsInt) else None


@attrs.frozen
class MSAGIssue:

    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_msag_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": "SrcMSAGRow", "dtype": "Int64"})
    """MSAG row number of the record if it is present in the MSAG, otherwise
    ``None``."""

    side: Literal["LEFT", "RIGHT"] | NoneOrNA = attrs.field(
        validator=in_(["LEFT", "RIGHT", None]),
        converter=_na_to_none,
        metadata={"table_field": "Side", "dtype": "string"})
    add_from: int | NoneOrNA = attrs.field(
        converter=pipe(_na_to_none, _int_or_none),
        validator=opt(instof((int, None))),
        metadata={"table_field": "AddrFrom", "dtype": "Int64"})
    add_to: int | NoneOrNA = attrs.field(
        converter=pipe(_na_to_none, _int_or_none),
        validator=opt(instof(int)),
        metadata={"table_field": "AddrTo", "dtype": "Int64"})
    parity: Parity | NoneOrNA = attrs.field(
        converter=_na_to_none,
        metadata={"table_field": "Parity", "dtype": "string"})
    street_name: str | NoneOrNA = attrs.field(
        converter=_na_to_none,
        metadata={"table_field": cf.lgcyfulst.name, "dtype": "string"})
    community: str | NoneOrNA = attrs.field(
        converter=_na_to_none,
        metadata={"table_field": cf.msagcomm.name, "dtype": "string"})
    state: str | NoneOrNA = attrs.field(
        converter=_na_to_none,
        metadata={"table_field": "State", "dtype": "string"})
    esn: str | NoneOrNA = attrs.field(
        converter=_na_to_none,
        metadata={"table_field": "ESN", "dtype": "string"})
    esn_same_lr: bool | NoneOrNA = attrs.field(
        default=None,
        converter=_na_to_none,
        metadata={"table_field": "ESNSameLR", "dtype": "boolean"})
    _message: Optional[str | Callable[[Self], str]] = attrs.field(default=None)

    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),
        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),
        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),
        metadata={"table_field": "RELATE_STR", "dtype": "string"}
    )

    @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__)
        values_dict = attrs.asdict(self)
        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]:
        return {
            dn: val
            for (_, metadata), val
            in self.metadata_with_values.items()
            if (dn := metadata.get("table_field")) is not None
        }

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

    @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:
        row_number: int = row.name  # type: ignore
        return cls(
            None, row_number,
            side, row["add_from"], row["add_to"], row["parity"],
            row["$name"], row["msagcomm"], row["state"], row["esn"],
            message=message
        )

    @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
class TNIssue:
    ...

class NG911MSAGComparator(NG911Session):
    """

    """

    def __init__(self,
                 workspace: PathLike[str] | str,
                 respect_submit: bool,
                 use_edit_session: bool = False,
                 messenger: Optional[GPMessenger] = None,
                 destination: PathLike[str] | str | None = 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, **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"))