src/nominatim_db/tools/postcodes.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Functions for importing, updating and otherwise maintaining the table
   9 of artificial postcode centroids.
  10 """
  11 from typing import Optional, Tuple, Dict, TextIO
  12 from collections import defaultdict
  13 from pathlib import Path
  14 import csv
  15 import gzip
  16 import logging
  17 import json
  18 from math import isfinite
  19
  20 from psycopg import sql as pysql
  21
  22 from ..db.connection import connect, Connection, table_exists
  23 from ..utils.centroid import PointsCentroid
  24 from ..data.postcode_format import PostcodeFormatter, CountryPostcodeMatcher
  25 from ..tokenizer.base import AbstractAnalyzer, AbstractTokenizer
  26
  27 LOG = logging.getLogger()
  28
  29
  30 def _to_float(numstr: str, max_value: float) -> float:
  31     """ Convert the number in string into a float. The number is expected
  32         to be in the range of [-max_value, max_value]. Otherwise rises a
  33         ValueError.
  34     """
  35     num = float(numstr)
  36     if not isfinite(num) or num <= -max_value or num >= max_value:
  37         raise ValueError()
  38
  39     return num
  40
  41
  42 def _extent_to_rank(extent: int) -> int:
  43     """ Guess a suitable search rank from the extent of a postcode.
  44     """
  45     if extent <= 100:
  46         return 25
  47     if extent <= 3000:
  48         return 23
  49     return 21
  50
  51
  52 def _open_external_file(fname: Path) -> TextIO:
  53     """ Open an external postcode data file, handling both flat text
  54         and gzip compression transparently based on the file extension.
  55     """
  56     if fname.suffix == '.gz':
  57         return gzip.open(fname, 'rt', encoding='utf-8')
  58     return open(fname, 'r', encoding='utf-8')
  59
  60
  61 class _PostcodeCollector:
  62     """ Collector for postcodes of a single country.
  63     """
  64
  65     def __init__(self, country: str, matcher: Optional[CountryPostcodeMatcher],
  66                  default_extent: int, exclude: set[str] = set()):
  67         self.country = country
  68         self.matcher = matcher
  69         self.extent = default_extent
  70         self.exclude = exclude
  71         self.collected: Dict[str, PointsCentroid] = defaultdict(PointsCentroid)
  72         self.normalization_cache: Optional[Tuple[str, Optional[str]]] = None
  73
  74     def add(self, postcode: str, x: float, y: float) -> None:
  75         """ Add the given postcode to the collection cache. If the postcode
  76             already existed, it is overwritten with the new centroid.
  77         """
  78         if self.matcher is not None:
  79             normalized: Optional[str]
  80             if self.normalization_cache and self.normalization_cache[0] == postcode:
  81                 normalized = self.normalization_cache[1]
  82             else:
  83                 match = self.matcher.match(postcode)
  84                 normalized = self.matcher.normalize(match) if match else None
  85                 self.normalization_cache = (postcode, normalized)
  86
  87             if normalized and normalized not in self.exclude:
  88                 self.collected[normalized] += (x, y)
  89
  90     def commit(self, conn: Connection, analyzer: AbstractAnalyzer,
  91                project_dir: Optional[Path], is_initial: bool) -> None:
  92         """ Update postcodes for the country from the postcodes selected so far.
  93
  94             When 'project_dir' is set, then any postcode files found in this
  95             directory are taken into account as well.
  96         """
  97         if project_dir is not None:
  98             self._update_from_external(analyzer, project_dir)
  99
 100         if is_initial:
 101             to_delete = []
 102         else:
 103             with conn.cursor() as cur:
 104                 # Ensure we only drop non-area artificial points on point updates
 105                 cur.execute("""SELECT postcode FROM location_postcodes
 106                                WHERE country_code = %s AND osm_id IS NULL AND NOT is_area""",
 107                             (self.country, ))
 108                 to_delete = [row[0] for row in cur if row[0] not in self.collected]
 109
 110         to_add = [dict(zip(('pc', 'x', 'y'), (k, *v.centroid())))
 111                   for k, v in self.collected.items()]
 112         self.collected = defaultdict(PointsCentroid)
 113
 114         LOG.info("Processing country '%s' (%s added, %s deleted).",
 115                  self.country, len(to_add), len(to_delete))
 116
 117         with conn.cursor() as cur:
 118             if to_add:
 119                 columns = ['country_code',
 120                            'rank_search',
 121                            'postcode',
 122                            'centroid',
 123                            'geometry']
 124                 values = [pysql.Literal(self.country),
 125                           pysql.Literal(_extent_to_rank(self.extent)),
 126                           pysql.Placeholder('pc'),
 127                           pysql.SQL('ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326)'),
 128                           pysql.SQL("""expand_by_meters(
 129                                            ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326), {})""")
 130                                .format(pysql.Literal(self.extent))]
 131                 if is_initial:
 132                     columns.extend(('place_id', 'indexed_status'))
 133                     values.extend((pysql.SQL("nextval('seq_place')"), pysql.Literal(1)))
 134
 135                 cur.executemany(pysql.SQL("INSERT INTO location_postcodes ({}) VALUES ({})")
 136                                      .format(pysql.SQL(',')
 137                                                   .join(pysql.Identifier(c) for c in columns),
 138                                              pysql.SQL(',').join(values)),
 139                                 to_add)
 140             if to_delete:
 141                 cur.execute("""DELETE FROM location_postcodes
 142                                WHERE country_code = %s and postcode = any(%s)
 143                                      AND osm_id is null and not is_area
 144                             """, (self.country, to_delete))
 145
 146     def _update_from_external(self, analyzer: AbstractAnalyzer, project_dir: Path) -> None:
 147         """ Look for an external postcode file for the active country in
 148             the project directory and add missing postcodes when found.
 149         """
 150         fname = self._find_external_centroid_file(project_dir)
 151         if fname is None:
 152             return
 153
 154         with _open_external_file(fname) as csvfile:
 155             reader = csv.DictReader(csvfile)
 156             for row in reader:
 157                 if 'postcode' not in row or 'lat' not in row or 'lon' not in row:
 158                     LOG.warning("Bad format for external postcode file for country '%s'."
 159                                 " Ignored.", self.country)
 160                     return
 161                 postcode = analyzer.normalize_postcode(row['postcode'])
 162                 if postcode not in self.collected and postcode not in self.exclude:
 163                     try:
 164                         # Do float conversation separately, it might throw
 165                         centroid = (_to_float(row['lon'], 180),
 166                                     _to_float(row['lat'], 90))
 167                         self.collected[postcode] += centroid
 168                     except ValueError:
 169                         LOG.warning("Bad coordinates %s, %s in '%s' country postcode file.",
 170                                     row['lat'], row['lon'], self.country)
 171
 172     def _find_external_centroid_file(self, project_dir: Path) -> Optional[Path]:
 173         for ext in ('csv', 'csv.gz'):
 174             fname = project_dir / f'{self.country}_postcodes.{ext}'
 175             if fname.is_file():
 176                 LOG.info("Using external postcode file '%s'.", fname)
 177                 return fname
 178         return None
 179
 180
 181 def update_postcodes(dsn: str, project_dir: Optional[Path],
 182                      tokenizer: AbstractTokenizer, force_reimport: bool = False) -> None:
 183     """ Update the table of postcodes from the input tables
 184         placex and place_postcode.
 185     """
 186     matcher = PostcodeFormatter()
 187     with tokenizer.name_analyzer() as analyzer:
 188         with connect(dsn) as conn:
 189             # Backfill country_code column where required
 190             conn.execute("""UPDATE place_postcode
 191                               SET country_code = get_country_code(centroid)
 192                               WHERE country_code is null
 193                          """)
 194             if force_reimport:
 195                 conn.execute("TRUNCATE location_postcodes")
 196                 is_initial = True
 197             else:
 198                 is_initial = _is_postcode_table_empty(conn)
 199             if is_initial:
 200                 conn.execute("""ALTER TABLE location_postcodes
 201                                 DISABLE TRIGGER location_postcodes_before_insert""")
 202             # Now update first postcode areas
 203             _update_postcode_areas(conn, analyzer, matcher, is_initial)
 204             # Update postcode areas from external geometry files
 205             _update_external_postcode_areas(conn, analyzer, matcher, project_dir, is_initial)
 206             # Then fill with estimated postcode centroids from other info
 207             _update_guessed_postcode(conn, analyzer, matcher, project_dir, is_initial)
 208             if is_initial:
 209                 conn.execute("""ALTER TABLE location_postcodes
 210                                 ENABLE TRIGGER location_postcodes_before_insert""")
 211             conn.commit()
 212
 213         analyzer.update_postcodes_from_db()
 214
 215
 216 def _is_postcode_table_empty(conn: Connection) -> bool:
 217     """ Check if there are any entries in the location_postcodes table yet.
 218     """
 219     with conn.cursor() as cur:
 220         cur.execute("SELECT place_id FROM location_postcodes LIMIT 1")
 221         return cur.fetchone() is None
 222
 223
 224 def _insert_postcode_areas(conn: Connection, country_code: str,
 225                            extent: int, pcs: list[dict[str, str]],
 226                            is_initial: bool) -> None:
 227     if pcs:
 228         with conn.cursor() as cur:
 229             columns = ['osm_id', 'country_code',
 230                        'rank_search', 'postcode', 'is_area',
 231                        'centroid', 'geometry']
 232             values = [pysql.Identifier('osm_id'), pysql.Identifier('country_code'),
 233                       pysql.Literal(_extent_to_rank(extent)), pysql.Placeholder('out'),
 234                       pysql.Literal(True), pysql.Identifier('centroid'),
 235                       pysql.Identifier('geometry')]
 236             if is_initial:
 237                 columns.extend(('place_id', 'indexed_status'))
 238                 values.extend((pysql.SQL("nextval('seq_place')"), pysql.Literal(1)))
 239
 240             cur.executemany(
 241                 pysql.SQL(
 242                     """ INSERT INTO location_postcodes ({})
 243                             SELECT {} FROM place_postcode
 244                             WHERE osm_type = 'R'
 245                                   and country_code = {} and postcode = %(in)s
 246                                   and geometry is not null
 247                     """).format(pysql.SQL(',')
 248                                      .join(pysql.Identifier(c) for c in columns),
 249                                 pysql.SQL(',').join(values),
 250                                 pysql.Literal(country_code)),
 251                 pcs)
 252
 253
 254 def _update_postcode_areas(conn: Connection, analyzer: AbstractAnalyzer,
 255                            matcher: PostcodeFormatter, is_initial: bool) -> None:
 256     """ Update the postcode areas made from postcode boundaries.
 257     """
 258     # first delete all areas that have gone
 259     if not is_initial:
 260         conn.execute(""" DELETE FROM location_postcodes pc
 261                          WHERE pc.osm_id is not null
 262                            AND NOT EXISTS(
 263                                   SELECT * FROM place_postcode pp
 264                                   WHERE pp.osm_type = 'R' and pp.osm_id = pc.osm_id
 265                                         and geometry is not null)
 266                     """)
 267     # now insert all in country batches, triggers will ensure proper updates
 268     with conn.cursor() as cur:
 269         cur.execute(""" SELECT country_code, postcode FROM place_postcode
 270                         WHERE geometry is not null and osm_type = 'R'
 271                         ORDER BY country_code
 272                     """)
 273         country_code = None
 274         fmt = None
 275         pcs = []
 276         for cc, postcode in cur:
 277             if country_code is None:
 278                 country_code = cc
 279                 fmt = matcher.get_matcher(country_code)
 280             elif country_code != cc:
 281                 _insert_postcode_areas(conn, country_code,
 282                                        matcher.get_postcode_extent(country_code), pcs,
 283                                        is_initial)
 284                 country_code = cc
 285                 fmt = matcher.get_matcher(country_code)
 286                 pcs = []
 287
 288             if fmt is not None:
 289                 if (m := fmt.match(postcode)):
 290                     pcs.append({'out': fmt.normalize(m), 'in': postcode})
 291
 292         if country_code is not None and pcs:
 293             _insert_postcode_areas(conn, country_code,
 294                                    matcher.get_postcode_extent(country_code), pcs,
 295                                    is_initial)
 296
 297
 298 def _find_external_geometry_file(project_dir: Path, country: str) -> Optional[Path]:
 299     for ext in ('jsonl', 'jsonl.gz'):
 300         fname = project_dir / f'{country}_postcodes_geometry.{ext}'
 301         if fname.is_file():
 302             LOG.info("Using external postcode file '%s'.", fname)
 303             return fname
 304     return None
 305
 306
 307 def _insert_external_postcode_areas(conn: Connection, country_code: str,
 308                                     extent: int, pcs: list[dict[str, Optional[str | float]]],
 309                                     is_initial: bool) -> None:
 310     if not pcs:
 311         return
 312
 313     if not is_initial:
 314         # first get the list of existing postcodes from previous geometry import.
 315         with conn.cursor() as cur:
 316             cur.execute("""SELECT postcode FROM location_postcodes
 317                             WHERE country_code = %s AND osm_id IS NULL AND is_area""",
 318                         (country_code,))
 319             existing_pcs = {row[0] for row in cur}
 320
 321         new_pcs = {p['postcode'] for p in pcs}
 322
 323         # Delete postcodes that were previously imported but no longer appear
 324         to_delete = existing_pcs - new_pcs
 325
 326         if to_delete:
 327             with conn.cursor() as cur:
 328                 cur.execute("""DELETE FROM location_postcodes
 329                             WHERE country_code = %s AND osm_id IS NULL AND is_area
 330                             AND postcode = any(%s)
 331                             """, (country_code, list(to_delete)))
 332
 333     with conn.cursor() as cur:
 334         columns = ['country_code', 'rank_search', 'postcode',
 335                    'is_area', 'centroid', 'geometry']
 336         values = [
 337             pysql.Literal(country_code),
 338             pysql.Literal(_extent_to_rank(extent)),
 339             pysql.Placeholder('postcode'),
 340             pysql.Literal(True),
 341             pysql.SQL("""COALESCE(
 342                            ST_SetSRID(ST_MakePoint(%(lon)s, %(lat)s), 4326),
 343                            ST_Centroid(ST_GeomFromGeoJSON(%(geometry)s))
 344                          )"""),
 345             pysql.SQL("ST_GeomFromGeoJSON(%(geometry)s)"),
 346         ]
 347         if is_initial:
 348             columns.extend(('place_id', 'indexed_status'))
 349             values.extend((pysql.SQL("nextval('seq_place')"), pysql.Literal(1)))
 350
 351         cur.executemany(
 352             pysql.SQL("INSERT INTO location_postcodes ({}) VALUES ({})").format(
 353                 pysql.SQL(',').join(pysql.Identifier(c) for c in columns),
 354                 pysql.SQL(',').join(values)
 355             ),
 356             pcs
 357         )
 358
 359
 360 def _update_external_postcode_areas(conn: Connection, analyzer: AbstractAnalyzer,
 361                                     matcher: PostcodeFormatter,
 362                                     project_dir: Optional[Path], is_initial: bool) -> None:
 363     if project_dir is None:
 364         return
 365
 366     with conn.cursor() as cur:
 367         cur.execute("SELECT country_code FROM country_name")
 368         todo_countries = {row[0] for row in cur}
 369
 370     # Exclude postcodes that are already covered by OSM areas.
 371     area_pcs: dict[str, set[str]] = defaultdict(set)
 372     with conn.cursor() as cur:
 373         cur.execute("""SELECT country_code, postcode FROM location_postcodes
 374                        WHERE is_area = true AND osm_id IS NOT NULL""")
 375         for cc, pc in cur:
 376             area_pcs[cc].add(pc)
 377
 378     for country in todo_countries:
 379         fmt = matcher.get_matcher(country)
 380         if fmt is None:
 381             continue
 382
 383         fname = _find_external_geometry_file(project_dir, country)
 384         if fname is None:
 385             continue
 386
 387         pcs = []
 388         with _open_external_file(fname) as jsonlfile:
 389             for i, line in enumerate(jsonlfile, start=1):
 390                 try:
 391                     data = json.loads(line)
 392                 except json.JSONDecodeError:
 393                     LOG.warning("Ignored line %s: bad JSON for country '%s'.", i, country)
 394                     continue
 395
 396                 geometry = data.get('geometry')
 397                 if (not isinstance(geometry, dict) or
 398                         geometry.get('type') not in ('Polygon', 'MultiPolygon') or
 399                         not isinstance(geometry.get('coordinates'), list)):
 400                     LOG.warning("Ignored line %s: bad geometry for country '%s'.", i, country)
 401                     continue
 402
 403                 props = data.get('properties')
 404                 if not isinstance(props, dict):
 405                     LOG.warning("Ignored line %s: bad properties for country '%s'.", i, country)
 406                     continue
 407
 408                 raw_postcode = props.get('postcode')
 409                 if not raw_postcode:
 410                     LOG.warning("Ignored line %s: missing postcode for country '%s'.", i, country)
 411                     continue
 412
 413                 m = fmt.match(raw_postcode)
 414                 if not m:
 415                     continue
 416                 normalized = fmt.normalize(m)
 417
 418                 if normalized in area_pcs[country]:
 419                     continue
 420
 421                 # parse optional centroid
 422                 lon, lat = None, None
 423                 if props.get('lon') is not None and props.get('lat') is not None:
 424                     try:
 425                         lat = _to_float(props['lat'], 90)
 426                         lon = _to_float(props['lon'], 180)
 427                     except ValueError:
 428                         LOG.warning("Bad centroid on line %s for country '%s', "
 429                                     "will use geometry centroid.", i, country)
 430                         lat, lon = None, None
 431
 432                 pcs.append({
 433                     'postcode': normalized,
 434                     'geometry': json.dumps(geometry),
 435                     'lon': lon,
 436                     'lat': lat,
 437                 })
 438
 439         _insert_external_postcode_areas(conn, country,
 440                                         matcher.get_postcode_extent(country),
 441                                         pcs, is_initial)
 442
 443
 444 def _update_guessed_postcode(conn: Connection, analyzer: AbstractAnalyzer,
 445                              matcher: PostcodeFormatter, project_dir: Optional[Path],
 446                              is_initial: bool) -> None:
 447     """ Computes artificial postcode centroids from the placex table,
 448         potentially enhances it with external postcode centroid data and then updates
 449         the postcodes in the table 'location_postcodes'.
 450     """
 451     # First get the list of countries that currently have postcodes.
 452     # (Doing this before starting to insert, so it is fast on import.)
 453     if is_initial:
 454         todo_countries: set[str] = set()
 455     else:
 456         with conn.cursor() as cur:
 457             cur.execute("""SELECT DISTINCT country_code FROM location_postcodes
 458                             WHERE osm_id is null AND not is_area""")
 459             todo_countries = {row[0] for row in cur}
 460
 461     # Next, get the list of postcodes already covered by areas (both OSM and geometry import).
 462     area_pcs = defaultdict(set)
 463     with conn.cursor() as cur:
 464         cur.execute("""SELECT country_code, postcode
 465                        FROM location_postcodes WHERE is_area
 466                        ORDER BY country_code""")
 467         for cc, pc in cur:
 468             area_pcs[cc].add(pc)
 469
 470     # Create a temporary table which contains coverage of the postcode areas imported from osm
 471     # and external geometry imported through xx_postcode_areas.jsonl.
 472     with conn.cursor() as cur:
 473         cur.execute("DROP TABLE IF EXISTS _global_postcode_area")
 474         cur.execute("""CREATE TABLE _global_postcode_area AS
 475                        (SELECT ST_SubDivide(ST_SimplifyPreserveTopology(
 476                                               ST_Union(geometry), 0.00001), 128) as geometry
 477                         FROM location_postcodes WHERE is_area)
 478                     """)
 479         cur.execute("CREATE INDEX ON _global_postcode_area USING gist(geometry)")
 480
 481     # Recompute the list of valid postcodes from placex.
 482     with conn.cursor(name="placex_postcodes") as cur:
 483         cur.execute("""
 484             SELECT country_code, postcode, ST_X(centroid), ST_Y(centroid)
 485               FROM (
 486                 (SELECT country_code, address->'postcode' as postcode, centroid
 487                   FROM placex WHERE address ? 'postcode')
 488                 UNION
 489                 (SELECT country_code, postcode, centroid
 490                  FROM place_postcode WHERE geometry is null)
 491               ) x
 492               WHERE not postcode like '%,%' and not postcode like '%;%'
 493                     AND NOT EXISTS(SELECT * FROM _global_postcode_area g
 494                                    WHERE ST_Intersects(x.centroid, g.geometry))
 495               ORDER BY country_code""")
 496
 497         collector = None
 498
 499         for country, postcode, x, y in cur:
 500             if collector is None or country != collector.country:
 501                 if collector is not None:
 502                     collector.commit(conn, analyzer, project_dir, is_initial)
 503                 collector = _PostcodeCollector(country, matcher.get_matcher(country),
 504                                                matcher.get_postcode_extent(country),
 505                                                exclude=area_pcs[country])
 506                 todo_countries.discard(country)
 507             collector.add(postcode, x, y)
 508
 509         if collector is not None:
 510             collector.commit(conn, analyzer, project_dir, is_initial)
 511
 512     # Now handle any countries that are only in the postcode table.
 513     for country in todo_countries:
 514         fmt = matcher.get_matcher(country)
 515         ext = matcher.get_postcode_extent(country)
 516         _PostcodeCollector(country, fmt, ext,
 517                            exclude=area_pcs[country]).commit(conn, analyzer, project_dir, False)
 518
 519     conn.execute("DROP TABLE IF EXISTS _global_postcode_area")
 520
 521
 522 def can_compute(dsn: str) -> bool:
 523     """ Check that the necessary tables exist so that postcodes can be computed.
 524     """
 525     with connect(dsn) as conn:
 526         return table_exists(conn, 'place_postcode')