Spaces:
Running
Running
| """DEV-ONLY offline geocoder for REDcert's 'Map by location' view. | |
| Mirrors scripts/geocode_holders.py (the ISCC geocoder) but sources REDcert's clean, already-split | |
| `city` / `postcode` / `country` fields — so no address-string parsing is needed. Writes | |
| `data/redcert/holder_coords.json` as {holder_norm: {CC: [lat, lng]}} (per-country, so a holder with | |
| plants in several countries is placed in each, never collapsed to one point). The app loads that file | |
| at runtime and plots stored points — NO live geocoding, so the map stays Streamlit-in-Snowflake safe. | |
| Strategy (best-effort, honest coverage): postcode via pgeocode (great for EU), then city via | |
| geonamescache, else skip (never fake a country-centroid point). | |
| Run: APP_BACKEND=local python scripts/geocode_redcert.py | |
| Deps (dev only): pip install pgeocode geonamescache | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import math | |
| import sys | |
| from collections import Counter | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) | |
| import redcert # noqa: E402 | |
| def _normalise_postcode(country: str, postcode: str) -> str: | |
| """Return the postcode shape expected by the country geocoder.""" | |
| pc = str(postcode or "").strip() | |
| if (country or "").upper() == "DE" and pc.isdigit() and len(pc) < 5: | |
| return pc.zfill(5) | |
| return pc | |
| def main() -> None: | |
| import geonamescache | |
| import pgeocode | |
| certs = redcert.load_redcert_certs() | |
| # Group the (postcode, city) addresses per (holder_norm, COUNTRY) — NOT per holder — so a holder | |
| # registered in several countries is geocoded once per country and never lands abroad. | |
| groups: dict = {} # (hn, cc) -> [(postcode, city)] | |
| for c in certs: | |
| hn = c.get("holder_norm") | |
| cc = (c.get("country") or "").upper() | |
| if not hn or not cc: | |
| continue | |
| addr = (str(c.get("postcode", "") or "").strip(), (c.get("city", "") or "").strip()) | |
| if not any(addr): | |
| continue | |
| addrs = groups.setdefault((hn, cc), []) | |
| if addr not in addrs: | |
| addrs.append(addr) | |
| gc = geonamescache.GeonamesCache() | |
| city_idx, pop = {}, {} | |
| for cd in gc.get_cities().values(): | |
| key = (cd["countrycode"], cd["name"].lower()) | |
| if key not in city_idx or cd["population"] > pop.get(key, 0): | |
| city_idx[key] = (round(cd["latitude"], 4), round(cd["longitude"], 4)) | |
| pop[key] = cd["population"] | |
| nomi_cache: dict = {} | |
| def nomi(cc: str): | |
| if cc not in nomi_cache: | |
| try: | |
| nomi_cache[cc] = pgeocode.Nominatim(cc) | |
| except Exception: | |
| nomi_cache[cc] = None | |
| return nomi_cache[cc] | |
| def geocode_one(cc: str, addrs: list): | |
| """Resolve the first address that yields a point, constrained to country `cc`: postcode via | |
| pgeocode(cc), then city via the (cc, city) index. Returns (latlng, how).""" | |
| for pc, city in addrs: | |
| pc = _normalise_postcode(cc, pc) | |
| if pc: | |
| n = nomi(cc) | |
| if n is not None: | |
| try: | |
| r = n.query_postal_code(pc) | |
| lat = float(r["latitude"]) | |
| if not math.isnan(lat): | |
| return (round(lat, 4), round(float(r["longitude"]), 4)), "postcode" | |
| except Exception: | |
| pass | |
| if city: | |
| hit = city_idx.get((cc, city.lower())) | |
| if hit: | |
| return hit, "city" | |
| return None, None | |
| coords: dict = {} # hn -> {CC: [lat, lng]} (per-country) | |
| src = Counter() | |
| pairs_done = 0 | |
| for (hn, cc), addrs in groups.items(): | |
| latlng, how = geocode_one(cc, addrs) | |
| if latlng: | |
| coords.setdefault(hn, {})[cc] = list(latlng) | |
| src[how] += 1 | |
| pairs_done += 1 | |
| out = Path(__file__).resolve().parent.parent / "data" / "redcert" / "holder_coords.json" | |
| out.parent.mkdir(parents=True, exist_ok=True) | |
| out.write_text(json.dumps(coords), encoding="utf-8") | |
| multi = sum(1 for v in coords.values() if len(v) > 1) | |
| n_holders = len({hn for hn, _ in groups}) | |
| print(f"geocoded {pairs_done} of {len(groups)} (holder,country) pairs " | |
| f"across {len(coords)} of {n_holders} holders — {dict(src)}") | |
| print(f" {multi} holders placed in >1 country (per-country shape)") | |
| print(f"wrote {out}") | |
| if __name__ == "__main__": | |
| main() | |