osm-import-tools/map-checker/check.py

#!/usr/bin/env python3
"""
map-checker: compare hand-recorded observations from thevillages.com/map
(see reference.yml) against our county/OSM address data.

This is a deliberately manual workflow. thevillages.com's Terms of Use
prohibit automated/bot access to their site, so nothing here fetches from
them - reference.yml is filled in BY HAND after a human looks at the
official map in their own browser. This script only reads OUR OWN data
(county-addresses.geojson / city-comparison.geojson, already produced by
convert-addresses.py and generate-city-report.py) and checks it against
those manually recorded entries.

Usage:
    python check.py [--reference reference.yml] [--data-dir /data] [--radius 200]
"""

import argparse
import json
import math
import re
import sys
from pathlib import Path

import yaml

DEFAULT_RADIUS_METERS = 200.0


def normalize_city(value):
    if not value:
        return ''
    s = str(value).strip()
    return '' if s in ('', 'nan', 'None') else s


def normalize_street(value):
    if not value:
        return ''
    s = str(value).strip().lower()
    s = re.sub(r'\bstate road\b', 'sr', s)
    s = re.sub(r'\bstate route\b', 'sr', s)
    s = re.sub(r'\bcounty road\b', 'cr', s)
    s = re.sub(r'\s+', ' ', s).strip()
    return s


def haversine_meters(lat1, lon1, lat2, lon2):
    r = 6371000.0
    p1, p2 = math.radians(lat1), math.radians(lat2)
    dp = math.radians(lat2 - lat1)
    dl = math.radians(lon2 - lon1)
    a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
    return 2 * r * math.asin(math.sqrt(a))


def load_geojson(path):
    if not path.exists():
        return None
    with open(path) as f:
        return json.load(f)


def find_matches(entry, county_features, comparison_features, radius_m):
    """Return a list of matched feature-property dicts for one reference entry.

    Prefers city-comparison.geojson (has county_postal_community, county_city,
    AND osm_city together) when available; falls back to the raw
    county-addresses.geojson (county-only, no OSM side) otherwise.
    """
    source = comparison_features if comparison_features is not None else county_features
    if source is None:
        return []

    street = entry.get('street')
    housenumber = entry.get('housenumber')
    lat, lon = entry.get('lat'), entry.get('lon')

    matches = []
    for feature in source['features']:
        props = feature.get('properties', {})

        if lat is not None and lon is not None:
            coords = feature['geometry']['coordinates']
            dist = haversine_meters(lat, lon, coords[1], coords[0])
            if dist > radius_m:
                continue
        elif street:
            if normalize_street(props.get('addr:street')) != normalize_street(street):
                continue
            if housenumber is not None:
                try:
                    if int(props.get('addr:housenumber')) != int(housenumber):
                        continue
                except (TypeError, ValueError):
                    continue
        else:
            continue

        matches.append(props)

    return matches


def check_entry(entry, index, county_features, comparison_features, radius_m):
    label = entry.get('street') or f"({entry.get('lat')}, {entry.get('lon')})"
    official_city = normalize_city(entry.get('official_city'))
    matches = find_matches(entry, county_features, comparison_features, radius_m)

    print(f"\n[{index}] {label}"
          + (f" #{entry['housenumber']}" if entry.get('housenumber') else '')
          + f"  ({entry.get('county', '?')})")
    if entry.get('official_village'):
        print(f"    Official village: {entry['official_village']}")
    print(f"    Official city (recorded): {official_city or '(none recorded)'}")

    if not matches:
        print("    No matching addresses found in our data - can't check.")
        return {'checked': True, 'matched': False, 'mismatches': 0}

    # A single street address (e.g. a shopping plaza or apartment building)
    # can match many unit rows that all carry identical city values - group
    # those together instead of printing one line per unit.
    groups = {}  # (postal_community, city, osm_city) -> {'count', 'addr'}
    for props in matches:
        addr = ' '.join(str(v) for v in
                         [props.get('addr:housenumber'), props.get('addr:street')] if v)
        key = (normalize_city(props.get('county_postal_community')),
               normalize_city(props.get('county_city')),
               normalize_city(props.get('osm_city')))
        if key not in groups:
            groups[key] = {'count': 0, 'addr': addr}
        groups[key]['count'] += 1

    mismatches = 0
    for (postal, city, osm), g in groups.items():
        fields = [('postal community', postal), ('city', city), ('OSM city', osm)]
        parts = []
        row_mismatch = False
        for field_label, value in fields:
            if not value:
                continue
            agrees = (not official_city) or value.lower() == official_city.lower()
            if not agrees:
                row_mismatch = True
            flag = '' if agrees else ' <-- MISMATCH'
            parts.append(f"{field_label}={value}{flag}")
        count_note = f" (x{g['count']})" if g['count'] > 1 else ''
        if row_mismatch:
            mismatches += g['count']
        print(f"    {g['addr']}{count_note}: " + ', '.join(parts))

    return {'checked': True, 'matched': True, 'mismatches': mismatches}


def main():
    parser = argparse.ArgumentParser(description=__doc__,
                                      formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--reference', default=str(Path(__file__).parent / 'reference.yml'))
    parser.add_argument('--data-dir', default='/data')
    parser.add_argument('--radius', type=float, default=DEFAULT_RADIUS_METERS,
                         help=f'Match radius in meters for lat/lon entries (default {DEFAULT_RADIUS_METERS})')
    args = parser.parse_args()

    ref_path = Path(args.reference)
    if not ref_path.exists():
        print(f"Error: {ref_path} not found.", file=sys.stderr)
        return 1

    with open(ref_path) as f:
        data = yaml.safe_load(f) or {}
    checks = data.get('checks') or []

    if not checks:
        print(f"No entries in {ref_path} yet.")
        print("Open https://www.thevillages.com/map/ in your own browser, look up an address,")
        print("and add an entry to reference.yml by hand - see the comments at the top of that file.")
        return 0

    data_dir = Path(args.data_dir)
    cache = {}
    total_mismatches = 0
    total_unmatched = 0

    for i, entry in enumerate(checks, 1):
        county = entry.get('county')
        if not county:
            print(f"\n[{i}] Skipped - missing 'county' field.")
            continue
        if county not in cache:
            cache[county] = (
                load_geojson(data_dir / county / 'county-addresses.geojson'),
                load_geojson(data_dir / county / 'city-comparison.geojson'),
            )
        county_features, comparison_features = cache[county]
        result = check_entry(entry, i, county_features, comparison_features, args.radius)
        if not result['matched']:
            total_unmatched += 1
        total_mismatches += result['mismatches']

    print(f"\n{'=' * 60}")
    print(f"Checked {len(checks)} entries: {total_mismatches} address(es) with a "
          f"disagreeing city field, {total_unmatched} entries with no match in our data.")

    return 0


if __name__ == '__main__':
    sys.exit(main())