mainegeo.matching

  1"""This module generates or loads a database of Maine town names and their
  2aliases, and provides a function for matching.
  3"""
  4
  5__docformat__ = 'google'
  6
  7__all__ = [
  8    'TownReference',
  9    'TownDatabase',
 10    'get_town_database'
 11]
 12
 13from dataclasses import dataclass
 14from functools import cached_property, cache
 15from typing import List, Dict, Optional, ClassVar
 16from pathlib import Path
 17from ruamel.yaml import YAML
 18from mainegeo.paths import TOWNSHIPS_JSON, TOWNSHIPS_YAML
 19from mainegeo.entities import (County, Cousub, TownType)
 20from mainegeo.townships import (
 21    clean_code,
 22    clean_town,
 23    strip_suffix,
 24    strip_region,
 25    toggle_suffix,
 26    extract_alias
 27)
 28
 29yaml = YAML()
 30
 31class MatchError(Exception):
 32    def __init__(self, message):
 33        super().__init__(message)
 34
 35@dataclass
 36class TownReference:
 37    name: str
 38    geocode: str
 39    gnis_id: int
 40    town_type: TownType
 41    county: County
 42    cousub: Cousub
 43    aliases: List[str]
 44    processed: Optional[bool] = False
 45
 46    def __post_init__(self):
 47        if not self.processed:
 48            self._clean_aliases()
 49            self._infer_aliases()
 50            self.processed = True
 51
 52    def _clean_aliases(self):
 53        aliases = sum(self.aliases, [])
 54        aliases = map(str.upper, filter(None, aliases))
 55        self.aliases = list(set(aliases))
 56        
 57    def _infer_aliases(self):
 58        aliases = self.aliases
 59        
 60        aliases.extend(list(map(clean_code, aliases)))
 61        aliases.extend(list(map(clean_town, aliases)))
 62        aliases.extend(list(map(strip_suffix, aliases)))
 63        aliases.extend(list(map(strip_region, aliases)))
 64        aliases.extend(list(map(strip_region, aliases))) # 2x
 65
 66        township_aliases = filter(None, map(extract_alias, aliases))
 67        aliases.extend(list(township_aliases))
 68
 69        if self.town_type in (TownType.UNORGANIZED, TownType.ISLAND):
 70            aliases.extend(list(map(toggle_suffix, aliases)))
 71        
 72        self.aliases = list(set(aliases))
 73        self.aliases.sort()
 74        
 75    @staticmethod
 76    def json_object_hook(json_record, processed = False):
 77        return TownReference(
 78            name = json_record['town'],
 79            geocode = json_record['town_geocode'],
 80            gnis_id = json_record['gnis_id'],
 81            town_type = TownType(json_record['geotype']),
 82            county = County(
 83                fips = json_record['county_fips'], 
 84                name = json_record['county_name'],
 85                code = json_record['sos_county']
 86            ),
 87            cousub = Cousub(
 88                geocode = json_record['cousub_geocode'],
 89                name = json_record['cousub_name'],
 90                basename = json_record['cousub_basename'],
 91                geoclass = json_record['classfp']
 92            ),
 93            aliases = [
 94                [
 95                    json_record['maine_gis_name'],
 96                    json_record['town'],
 97                    json_record['cousub_basename'],
 98                    json_record['voting_name'],
 99                    json_record['tribal_name']
100                ],
101                json_record['gnis_variants'],
102                json_record['historical_names'],
103                json_record['misspellings'],
104                json_record['islands']
105            ],
106            processed = processed
107        )
108        
109@dataclass(frozen=True)
110class TownAlias:
111    """ A lightweight frozen container holding the minimum elements required for matching.
112    """
113    name: str
114    county_fips: Optional[int] = None
115
116@dataclass
117class TownDatabase:
118    data: List[TownReference] = None
119    processed: bool = False
120    
121    YAML_PATH: ClassVar[Path] = TOWNSHIPS_YAML
122    JSON_PATH: ClassVar[Path] = TOWNSHIPS_JSON
123    
124    def __post_init__(self):
125        if self.processed is False:
126            self._process_data()
127            self._validate_data()
128
129    @classmethod
130    def build(cls):
131        file_path = cls.YAML_PATH    
132        # first call
133        if Path(file_path).exists():
134            return cls.load_from_yaml(file_path)
135        # subsequent calls
136        else:
137            towndb = cls.create_from_raw_data()
138            towndb.save_to_yaml(file_path)
139            return towndb
140
141    @classmethod
142    def create_from_raw_data(cls, json_path: Path = None):
143        import json
144        
145        if json_path is None:
146            json_path = cls.JSON_PATH
147            
148        with open(json_path) as file:
149            towns = json.load(file, object_hook = TownReference.json_object_hook)
150            return cls(towns, processed = False)
151    
152    @classmethod
153    def load_from_yaml(cls, file_path = None):
154        file_path = file_path or cls.YAML_PATH
155         
156        with open(file_path, 'r') as f:
157            data = yaml.load(f)
158            towns = []
159            
160            for yml in data['towns']:
161                town = TownReference(
162                    name = yml['name'],
163                    geocode = yml['geocode'],
164                    gnis_id = yml['gnis_id'],
165                    town_type = TownType(yml['town_type']),
166                    county = County(
167                        fips = yml['county']['fips'], 
168                        name = yml['county']['name'],
169                        code = yml['county']['code']
170                    ),
171                    cousub = Cousub(
172                        geocode = yml['cousub']['geocode'],
173                        name = yml['cousub']['name'],
174                        basename = yml['cousub']['basename'],
175                        geoclass = yml['cousub']['geoclass']
176                    ),
177                    aliases = yml['aliases'],
178                    processed = True
179                )
180                towns.append(town)
181            
182            return cls(towns, processed = True)
183        
184    def save_to_yaml(self, file_path = None):
185        file_path = file_path or self.YAML_PATH
186        
187        serializable_data = {
188            'towns': [
189                {
190                    'name': town.name,
191                    'geocode': town.geocode,
192                    'town_type': town.town_type.value,
193                    'gnis_id': town.gnis_id,
194                    'county': {
195                        'fips': town.county.fips,
196                        'name': town.county.name,
197                        'code': town.county.code
198                    },
199                    'cousub' : {
200                        'geocode': town.cousub.geocode,
201                        'name': town.cousub.name,
202                        'basename': town.cousub.basename,
203                        'geoclass': town.cousub.geoclass
204                    },
205                    'aliases': town.aliases
206                } for town in self.data
207            ]
208        }
209
210        file_path.parent.mkdir(parents = True, exist_ok = True)
211        
212        with open(file_path, 'w') as f:
213            yaml.dump(serializable_data, f)
214
215    def _process_data(self):
216        if self.data is not None:
217            self._remove_duplicate_aliases()
218            self.data.sort(key=lambda x: x.name)
219            self.processed = True
220
221    def _validate_data(self):
222        geocodes = [town.geocode for town in self.data]
223        if not all(geocodes):
224            raise ValueError("Missing geocodes in source data")
225        elif len(geocodes) != len(set(geocodes)):
226            raise ValueError("Non-unique geocodes in source data")
227
228    def _remove_duplicate_aliases(self):
229        for town in self.data:
230            town.aliases = [
231                alias_name for alias_name in town.aliases
232                if TownAlias(alias_name, town.county.fips) in self.alias_lookup.keys()
233            ]
234
235    @cached_property
236    def _suggested_aliases(self) -> List[TownAlias]:
237        all = []
238        for town in self.data:
239            for alias_name in town.aliases:
240                all.append(TownAlias(alias_name))
241                all.append(TownAlias(alias_name, town.county.fips))
242        return all
243    
244    @cached_property
245    def _unique_aliases(self) -> List[TownAlias]:
246        counts = {}
247        for alias in self._suggested_aliases:
248            counts[alias] = counts.get(alias, 0) + 1
249        return [alias for alias, count in counts.items() if count == 1]
250    
251    @cached_property
252    def alias_lookup(self) -> Dict[TownAlias, TownReference]:
253        records = {}
254        for town in self.data:
255            canonical = town.name.upper()
256            
257            for alias_name in town.aliases:
258                state_alias = TownAlias(alias_name)
259                county_alias = TownAlias(alias_name, town.county.fips)
260                
261                for alias in (state_alias, county_alias):
262                    if alias in self._unique_aliases or alias_name == canonical:
263                        records[alias] = town
264        return records
265            
266    def search(self, **kwargs) -> List[TownReference]:
267        return [
268            town for town in self.data
269            if all(getattr(town, k) == v for k, v in kwargs.items())
270        ]
271            
272    def match(
273        self,
274        town: str,
275        county_fips: int = None,
276        cleaned: bool = False,
277        strict: bool = False
278    ) -> TownReference:
279        """
280        Clean and match a town name to the alias database and return the `TownReference` object.
281        
282        For faster matches on already clean data, 
283        pass `cleaned = True` to match without cleaning.
284
285        Args:
286            town: A single town or township name
287            county_fips: Integer code for county. If used, will improve match rate.
288            cleaned: True if the town name is already clean, False if it should be cleaned.
289            strict: True if exception should be raised when a match fails
290
291        Examples:
292            >>> towndb = TownDatabase.build()
293            >>> towndb.match('Cross Lake Twp (T17 R5)').name
294            'Cross Lake Twp'
295            >>> towndb.match('Prentiss Twp', cleaned = True) is None
296            True
297            >>> towndb.match('Prentiss Twp', county_fips = 19, cleaned = True).name
298            'Prentiss Twp T7 R3 NBPP'
299        """
300        town_name = town.upper() if cleaned else clean_town(town.upper())
301
302        def lazy_get_code(unmatched_town: str):
303            township_code = clean_code(unmatched_town)
304            if township_code != unmatched_town:
305                return township_code
306            
307        def lazy_get_alias(unmatched_town: str):
308            township_alias = extract_alias(unmatched_town)
309            if township_alias != unmatched_town:
310                return township_alias
311        
312        # lazy evaluation
313        names = [
314            lambda: town_name,
315            lambda: lazy_get_code(town_name),
316            lambda: lazy_get_alias(town_name)
317        ]
318
319        for get_name in names:
320            name = get_name()
321            if name:           
322                state_alias = TownAlias(name)
323                state_match = self.alias_lookup.get(state_alias)
324                
325                if state_match:
326                    return state_match
327                
328                if county_fips is None:
329                    continue
330
331                county_alias = TownAlias(name, county_fips)
332                county_match = self.alias_lookup.get(county_alias)
333
334                if county_match:
335                    return county_match
336                
337        if strict:
338            message = f'No match found for {town}.'
339            raise MatchError(message = message)
340            
341    def canonical_name(self, town: str, county_fips: int = None, cleaned: bool = False) -> str:
342        """
343        Match a town to the alias database and return the canonical name only.
344        
345        Convenience method; equivalent to `TownDatabase.build().match(...).name`.
346        
347        For faster matches on already clean data, 
348        pass `cleaned = True` to match without cleaning.
349
350        Args:
351            town: A single town or township name
352            county_fips: Integer code for county. If used, will improve match rate.
353            cleaned: True if the town name is already clean, False if it should be cleaned.
354
355        Examples:
356            >>> towndb = TownDatabase.build()
357            >>> towndb.canonical_name('Cross Lake Twp (T17 R5)')
358            'Cross Lake Twp'
359            >>> towndb.canonical_name('Soldiertown Twp') is None
360            True
361            >>> towndb.canonical_name('Soldiertown Twp', county_fips = 25)
362            'Soldiertown Twp T2 R3 NBKP'
363        """
364        match = self.match(town, county_fips, cleaned = cleaned)
365        if match:
366            return match.name
367
368@cache        
369def get_town_database() -> TownDatabase:
370    return TownDatabase.build()