diff --git a/back-end/controllers/celery_controller/celery_config.py b/back-end/controllers/celery_controller/celery_config.py index ebdfd0a..1916101 100755 --- a/back-end/controllers/celery_controller/celery_config.py +++ b/back-end/controllers/celery_controller/celery_config.py @@ -28,7 +28,14 @@ def make_celery(): "sweep-stuck-tasks": { "task": "controllers.celery_controller.celery_tasks.sweep_stuck_tasks", "schedule": 300.0, - } + }, + # Edit retiles splice only the dirty region (z9-16); this settles + # the deliberately stale z0-8 overview tiles once a folder has + # been quiet for a while. + "settle-stale-tiles": { + "task": "controllers.celery_controller.celery_tasks.settle_stale_tiles", + "schedule": 300.0, + }, }, ) return celery diff --git a/back-end/controllers/celery_controller/celery_tasks.py b/back-end/controllers/celery_controller/celery_tasks.py index 5575813..64d578f 100755 --- a/back-end/controllers/celery_controller/celery_tasks.py +++ b/back-end/controllers/celery_controller/celery_tasks.py @@ -153,6 +153,10 @@ def process_data(self, folderid, operation): # fabric intake dispatches the recompute when the new fabric lands. has_fabric = bool(file_ops.get_files_by_type(folderid, "fabric", session)) + # The fabric is the same for every coverage file — parse it once for + # the whole run, not once per file (it's an ~80 MB CSV). + fabric_gdf = kml_ops.load_fabric_gdf(folderid) if has_fabric else None + for file in coverage_files: if not has_fabric: file.computed = False @@ -183,6 +187,7 @@ def process_data(self, folderid, operation): latency, category, session, + fabric=fabric_gdf, ) file.computed = True @@ -250,9 +255,12 @@ def async_delete_files(self, file_ids, editfile_ids): # with a dirty flag so rapid edits coalesce. TILES_DIRTY_KEY = "bdk:tiles-dirty:{folderid}" +TILES_DIRTY_BBOX_KEY = "bdk:tiles-dirty-bbox:{folderid}" +TILES_SETTLE_KEY = "bdk:tiles-settle:{folderid}" TILES_LOCK_KEY = "bdk:tiles-lock:{folderid}" TILES_LOCK_TTL = 3600 # safety expiry on the lock; no rebuild should take this long TILES_LOCK_WAIT = 1800 # max time a regenerate waits behind another rebuild +TILES_SETTLE_QUIET = 600 # settle the stale z0-8 only after edits go quiet this long def _tiles_redis(): @@ -269,10 +277,59 @@ def _tiles_redis(): return None -def _mark_tiles_dirty(folderid): +def _mark_tiles_dirty(folderid, bbox=None): + """Record that the folder's tiles no longer match DB truth. With a bbox + ([minx, miny, maxx, maxy] lon/lat) the dirt is scoped to that region and + the rebuild may splice just its tiles; without one the whole tileset is + stale. A bbox never downgrades already-recorded whole-tileset dirt.""" r = _tiles_redis() - if r is not None: - r.set(TILES_DIRTY_KEY.format(folderid=folderid), "1") + if r is None: + return + dirty_key = TILES_DIRTY_KEY.format(folderid=folderid) + if bbox is None: + r.set(dirty_key, "full") + else: + r.rpush(TILES_DIRTY_BBOX_KEY.format(folderid=folderid), json.dumps(bbox)) + r.set(dirty_key, "bbox", nx=True) + + +def _pop_tiles_dirt(r, folderid): + """Atomically consume the folder's recorded dirt -> (flag, [bbox, ...]). + Atomic so a mark racing this pop is either fully consumed now or fully + left for the next rebuild — never half-eaten.""" + pipe = r.pipeline(transaction=True) + pipe.lrange(TILES_DIRTY_BBOX_KEY.format(folderid=folderid), 0, -1) + pipe.delete(TILES_DIRTY_BBOX_KEY.format(folderid=folderid)) + pipe.get(TILES_DIRTY_KEY.format(folderid=folderid)) + pipe.delete(TILES_DIRTY_KEY.format(folderid=folderid)) + raw_bboxes, _, dirty, _ = pipe.execute() + return dirty, [json.loads(b) for b in raw_bboxes] + + +def _features_bbox(features): + """[minx, miny, maxx, maxy] over every coordinate of the given GeoJSON + features, or None when there are no coordinates.""" + coords = [] + + def walk(node): + if isinstance(node, (list, tuple)): + if ( + len(node) >= 2 + and isinstance(node[0], (int, float)) + and isinstance(node[1], (int, float)) + ): + coords.append((node[0], node[1])) + else: + for item in node: + walk(item) + + for feature in features or []: + walk(((feature or {}).get("geometry") or {}).get("coordinates")) + if not coords: + return None + xs = [c[0] for c in coords] + ys = [c[1] for c in coords] + return [min(xs), min(ys), max(xs), max(ys)] @celery.task(bind=True, autoretry_for=(Exception,), retry_backoff=True) @@ -367,9 +424,14 @@ def apply_edit_changes(self, markers, folderid, polygonfeatures): + file_ops.get_files_with_postfix(user_folder.id, ".geojson", session) ] + from controllers.database_controller.setting_ops import export_max_service_only + results = session.query(kml_data).filter(kml_data.file_id.in_(all_file_ids)).all() availability_csv = kml_ops.generate_csv_data( - results, user_folder.organization.provider_id, user_folder.organization.brand_name + results, + user_folder.organization.provider_id, + user_folder.organization.brand_name, + max_service_only=export_max_service_only(session), ) csv_name = f"availability-{datetime.now().strftime('%Y-%m-%d_%H:%M:%S')}.csv" @@ -391,24 +453,16 @@ def apply_edit_changes(self, markers, folderid, polygonfeatures): session.commit() session.close() - _mark_tiles_dirty(folderid) + # An edit only changes points inside its drawn polygons, so the dirt is + # scoped: the rebuild can splice just that region's z9-16 tiles. + _mark_tiles_dirty(folderid, bbox=_features_bbox(polygonfeatures)) def _rebuild_folder_tiles(folderid): """Rebuild a folder's vector tiles from current DB truth (the slow part).""" session = Session() try: - geojson_data = [] - all_kmls = file_ops.get_files_with_postfix(folderid, ".kml", session) - for kml_f in all_kmls: - geojson_data.append(vt_ops.read_kml(kml_f.id, session)) - - all_geojsons = file_ops.get_files_with_postfix( - folderid=folderid, postfix=".geojson", session=session - ) - for geojson_f in all_geojsons: - geojson_data.append(vt_ops.read_geojson(geojson_f.id, session)) - + geojson_data = vt_ops.folder_coverage_features(folderid, session) mbtiles_ops.delete_mbtiles(folderid, session) vt_ops.create_tiles(geojson_data, folderid, session) finally: @@ -416,20 +470,38 @@ def _rebuild_folder_tiles(folderid): session.close() +def _splice_folder_tiles(folderid, bboxes): + """Splice the dirty regions' z9-16 tiles into the live tileset. True on + success; False (fall back to a full rebuild) on any failure.""" + try: + session = Session() + try: + return vt_ops.splice_tiles(folderid, bboxes, session) + finally: + session.close() + except Exception: + logger.exception(f"tile splice failed for folder {folderid}; doing a full rebuild") + return False + + def _coalesced_tile_rebuild(folderid, owner_id): - """Rebuild a folder's tiles, single-flight + coalescing. Callers mark the - folder dirty after changing its data; the rebuild that runs after that - clears the flag. A caller that finds the folder clean no-ops (an earlier - rebuild already covered its change), so N rapid changes cost ~1 rebuild. - Returns only once the tiles cover the caller's change. Without redis, the - debounce degrades to always rebuilding (correct, less efficient).""" + """Refresh a folder's tiles, single-flight + coalescing. Callers mark the + folder dirty after changing its data; the refresh that runs after that + consumes the dirt. A caller that finds the folder clean no-ops (an + earlier refresh already covered its change), so N rapid changes cost ~1 + refresh. Returns only once the tiles cover the caller's change. + + Dirt scoped to bboxes (edits) is SPLICED — only the dirty regions' z9-16 + tiles are regenerated, into the current tileset — leaving the z0-8 + overview tiles slightly stale (sub-pixel at those zooms); the settle + task full-rebuilds once the folder goes quiet. Whole-tileset dirt, a + splice failure, or no redis means a full rebuild.""" r = _tiles_redis() if r is None: _rebuild_folder_tiles(folderid) return "tiles rebuilt" lock_key = TILES_LOCK_KEY.format(folderid=folderid) - dirty_key = TILES_DIRTY_KEY.format(folderid=folderid) waited = 0 while not r.set(lock_key, str(owner_id), nx=True, ex=TILES_LOCK_TTL): if waited >= TILES_LOCK_WAIT: @@ -437,10 +509,14 @@ def _coalesced_tile_rebuild(folderid, owner_id): time.sleep(5) waited += 5 try: - if not r.get(dirty_key): + dirty, bboxes = _pop_tiles_dirt(r, folderid) + if not dirty: return "tiles fresh (an earlier rebuild covered this edit)" - r.delete(dirty_key) + if dirty == b"bbox" and bboxes and _splice_folder_tiles(folderid, bboxes): + r.set(TILES_SETTLE_KEY.format(folderid=folderid), str(time.time())) + return "tiles spliced" _rebuild_folder_tiles(folderid) + r.delete(TILES_SETTLE_KEY.format(folderid=folderid)) return "tiles rebuilt" finally: r.delete(lock_key) @@ -448,12 +524,42 @@ def _coalesced_tile_rebuild(folderid, owner_id): @celery.task(bind=True, autoretry_for=(Exception,), retry_backoff=True) def regenerate_tiles(self, folderid): - """The slow half of an edit chain: rebuild the folder's tiles via the - shared coalescing path. 'Task SUCCESS' always means 'the tiles include - this chain's edit'.""" + """The slow half of an edit chain: refresh the folder's tiles via the + shared coalescing path (a splice when the dirt is edit-scoped, a full + rebuild otherwise). 'Task SUCCESS' always means 'the tiles include this + chain's edit'.""" return _coalesced_tile_rebuild(folderid, self.request.id) +@celery.task +def settle_stale_tiles(): + """Beat housekeeping: splices leave a folder's z0-8 overview tiles + slightly stale (an edited dot is sub-pixel at those zooms). Once a + spliced folder has been quiet for TILES_SETTLE_QUIET, run one full + rebuild to reconcile. Background work — no job row, so no pill.""" + r = _tiles_redis() + if r is None: + return "no redis; nothing to settle" + settled = [] + for key in r.scan_iter(match=TILES_SETTLE_KEY.format(folderid="*")): + key = key.decode() if isinstance(key, bytes) else key + try: + folderid = int(key.rsplit(":", 1)[1]) + last_splice = float(r.get(key) or 0) + except (ValueError, AttributeError): + r.delete(key) + continue + if time.time() - last_splice < TILES_SETTLE_QUIET: + continue + # Consume the flag first: if the rebuild fails the next edit's splice + # re-flags, and the dirty mark below survives for the retry anyway. + r.delete(key) + _mark_tiles_dirty(folderid) + regenerate_tiles.apply_async(args=[folderid]) + settled.append(folderid) + return f"settling folders {settled}" if settled else "nothing to settle" + + @celery.task(bind=True, autoretry_for=(Exception,), retry_backoff=True) def async_folder_copy_for_export(self, folderid, serialized_csv, brandname, deadline): from services import export_service @@ -678,10 +784,10 @@ def raster2vector(self, data, userid, outfile_name): geojson_array = [] all_kmls = file_ops.get_files_with_postfix(fileVal.folder_id, ".kml", session) for kml_f in all_kmls: - geojson_array.append(vt_ops.read_kml(kml_f.id, session)) + geojson_array.extend(vt_ops.read_kml(kml_f.id, session)) all_geojsons = file_ops.get_files_with_postfix(fileVal.folder_id, ".geojson", session) for geojson_f in all_geojsons: - geojson_array.append(vt_ops.read_geojson(geojson_f.id, session)) + geojson_array.extend(vt_ops.read_geojson(geojson_f.id, session)) logger.info("Creating Vector Tiles") mbtiles_ops.delete_mbtiles(fileVal.folder_id, session) diff --git a/back-end/controllers/database_controller/file_ops.py b/back-end/controllers/database_controller/file_ops.py index 1ba986e..9460526 100755 --- a/back-end/controllers/database_controller/file_ops.py +++ b/back-end/controllers/database_controller/file_ops.py @@ -38,9 +38,13 @@ def get_files_with_postfix(folderid, postfix, session=None): owns_session = True try: + # Deterministic order: downstream merges (get_kml_data's per-file + # property stamping, the tile pipeline's feature stream) must produce + # identical output for identical inputs. files_with_ending = ( session.query(file) .filter(file.folder_id == folderid, file.name.endswith(postfix)) + .order_by(file.id) .all() ) return files_with_ending @@ -72,7 +76,10 @@ def get_files_by_type(folderid, filetype, session=None): try: files_with_type = ( - session.query(file).filter(file.folder_id == folderid, file.type == filetype).all() + session.query(file) + .filter(file.folder_id == folderid, file.type == filetype) + .order_by(file.id) + .all() ) return files_with_type except NoResultFound: diff --git a/back-end/controllers/database_controller/kml_ops.py b/back-end/controllers/database_controller/kml_ops.py index cd45e58..f998aa6 100755 --- a/back-end/controllers/database_controller/kml_ops.py +++ b/back-end/controllers/database_controller/kml_ops.py @@ -2,25 +2,20 @@ import json import logging from io import StringIO -from multiprocessing import Lock import geopandas import pandas -import shapely from shapely.geometry import shape -from sqlalchemy.exc import IntegrityError, SQLAlchemyError +from sqlalchemy.exc import SQLAlchemyError from database.models import fabric_data, kml_data from database.sessions import Session from utils.logger_config import logger -from utils.settings import BATCH_SIZE from .file_editfile_link_ops import get_editfiles_for_file from .file_ops import get_file_with_id, get_files_by_type, get_files_with_postfix from .geo_io import read_geo_bytes, suffix_for -db_lock = Lock() - def get_kml_data(folderid, session=None): owns_session = False @@ -39,6 +34,8 @@ def get_kml_data(folderid, session=None): all_data = {} if len(fabric_files) > 0: for fabric_file in fabric_files: + # Ordered so the tile pipeline's feature stream (and therefore + # tile bytes) is a deterministic function of the data. all_locations = ( session.query( fabric_data.location_id, @@ -48,8 +45,9 @@ def get_kml_data(folderid, session=None): fabric_data.bsl_flag, ) .filter(fabric_data.file_id == fabric_file.id) + .order_by(fabric_data.location_id) .all() - ) # Change to fabric_file.id + ) # Initialize a dictionary to hold location_id as key and its data as value, including location_id itself all_data.update( @@ -195,63 +193,93 @@ def get_kml_data_by_file(fileid, session=None): def add_to_db(pandaDF, kmlid, download, upload, tech, wireless, latency, category, session): - batch = [] - + """Write one coverage file's computed rows via COPY (one statement on the + session's own connection, so it commits with the session) — orders of + magnitude faster than per-row ORM objects for the big result sets.""" fileVal = get_file_with_id(kmlid) - for _, row in pandaDF.iterrows(): - try: - if row.location_id == "": - continue + try: + rows = pandaDF[pandaDF.location_id != ""] + if download == "": + download = 0 - if download == "": - download = 0 - - newData = kml_data( - location_id=int(row.location_id), - served=True, - wireless=wireless, - lte=False, - coveredLocations=fileVal.name, - maxDownloadNetwork=fileVal.name, - maxDownloadSpeed=int(download), - maxUploadSpeed=int(upload), - techType=tech, - file_id=fileVal.id, - address_primary=row.address_primary, - longitude=row.longitude, - latitude=row.latitude, - latency=latency, - category=category, + columns = [ + "location_id", + "served", + "wireless", + "lte", + "coveredLocations", + "maxDownloadNetwork", + "maxDownloadSpeed", + "maxUploadSpeed", + "techType", + "file_id", + "address_primary", + "longitude", + "latitude", + "latency", + "category", + ] + out = pandas.DataFrame( + { + "location_id": rows.location_id.astype(int).values, + "served": True, + "wireless": bool(wireless), + "lte": False, + "coveredLocations": fileVal.name, + "maxDownloadNetwork": fileVal.name, + "maxDownloadSpeed": int(download), + "maxUploadSpeed": int(upload), + "techType": tech, + "file_id": fileVal.id, + "address_primary": rows.address_primary.values, + "longitude": rows.longitude.values, + "latitude": rows.latitude.values, + "latency": latency, + "category": category, + }, + columns=columns, + ) + + if len(out): + buf = StringIO() + out.to_csv(buf, index=False, header=False, na_rep="\\N") + buf.seek(0) + cols_sql = ", ".join(f'"{c}"' for c in columns) + cursor = session.connection().connection.cursor() + cursor.copy_expert( + f"COPY kml_data ({cols_sql}) FROM STDIN WITH (FORMAT csv, NULL '\\N')", buf ) - batch.append(newData) - - if len(batch) >= BATCH_SIZE: - with db_lock: - try: - session.bulk_save_objects(batch) - session.commit() - except IntegrityError: - session.rollback() - - batch = [] - except Exception as e: - logging.error(f"Error occurred while inserting data: {e}") - return False - - if batch: - with db_lock: - try: - session.bulk_save_objects(batch) - session.commit() - except IntegrityError: - session.rollback() - session.commit() + session.commit() + except Exception as e: + session.rollback() + logging.error(f"Error occurred while inserting data: {e}") + return False return True -def generate_csv_data(results, provider_id, brand_name): +def filter_max_service(availability_csv): + """One BDC row per location: the fastest claim wins (download desc, then + upload desc, then low-latency first, then lowest technology code as a + deterministic tiebreak). The surviving row keeps its own values. The BDC + accepts multiple technology claims per location, so this is OPTIONAL + behavior behind the export_max_service_only site setting — default is to + report every claim.""" + ordered = availability_csv.sort_values( + by=[ + "max_advertised_download_speed", + "max_advertised_upload_speed", + "low_latency", + "technology", + ], + ascending=[False, False, False, True], + kind="mergesort", + ) + return ordered.drop_duplicates(subset=["location_id"], keep="first") + + +def generate_csv_data(results, provider_id, brand_name, max_service_only=False): availability_csv = pandas.DataFrame() availability_csv["location_id"] = [row.location_id for row in results] @@ -266,6 +294,8 @@ def generate_csv_data(results, provider_id, brand_name): availability_csv.drop_duplicates( subset=["location_id", "technology"], keep="first", inplace=True ) + if max_service_only: + availability_csv = filter_max_service(availability_csv) availability_csv = availability_csv[ [ "provider_id", @@ -294,7 +324,11 @@ def export(folderid, providerid, brandname, deadline, session, dispatch_copy=Tru all_file_ids = [file.id for file in all_files] results = session.query(kml_data).filter(kml_data.file_id.in_(all_file_ids)).all() - availability_csv = generate_csv_data(results, providerid, brandname) + from .setting_ops import export_max_service_only + + availability_csv = generate_csv_data( + results, providerid, brandname, max_service_only=export_max_service_only(session) + ) output = io.BytesIO() availability_csv.to_csv(output, index=False, encoding="utf-8") @@ -404,28 +438,35 @@ def reapply_plan_markers(coverage_file, session): session.commit() -def compute_wireless_locations(folderid, kmlid, download, upload, tech, latency, category, session): +def load_fabric_gdf(folderid): + """Parse a folder's active-fabric CSVs into ONE point GeoDataFrame. - # Only the active (BSL) fabric drives coverage — non_bsl and supplemental - # CSVs live in the folder too but never enter computation. + This is the heavy, per-folder-constant part of every coverage compute + (an ~80 MB CSV parse + point construction), so callers recomputing + several coverage files should load it once and pass it to + add_network_data instead of paying it per file. Only the active (BSL) + fabric drives coverage — non_bsl and supplemental CSVs live in the + folder too but never enter computation. Returns None with no fabric.""" fabric_files = get_files_by_type(folderid, "fabric") - coverage_file = get_file_with_id(kmlid) - - if fabric_files is None or coverage_file is None: - raise FileNotFoundError("Fabric or coverage file not found in the database") - - fabric_arr = [] - for fabric_file in fabric_files: - tempdf = pandas.read_csv(StringIO(fabric_file.data.decode())) - fabric_arr.append(tempdf) - df = pandas.concat(fabric_arr) - - fabric = geopandas.GeoDataFrame( + if not fabric_files: + return None + df = pandas.concat([pandas.read_csv(StringIO(ff.data.decode())) for ff in fabric_files]) + return geopandas.GeoDataFrame( df, crs="EPSG:4326", - geometry=[shapely.geometry.Point(xy) for xy in zip(df.longitude, df.latitude)], + geometry=geopandas.points_from_xy(df.longitude, df.latitude), ) + +def compute_wireless_locations( + folderid, kmlid, download, upload, tech, latency, category, session, fabric=None +): + coverage_file = get_file_with_id(kmlid) + if fabric is None: + fabric = load_fabric_gdf(folderid) + if fabric is None or coverage_file is None: + raise FileNotFoundError("Fabric or coverage file not found in the database") + wireless_coverage = read_geo_bytes(coverage_file.data, suffix_for(coverage_file.name)) wireless_coverage = wireless_coverage.to_crs("EPSG:4326") @@ -453,26 +494,13 @@ def compute_wireless_locations(folderid, kmlid, download, upload, tech, latency, def preview_wireless_locations(folderid, kml_filename): - - fabric_files = get_files_by_type(folderid, "fabric") with open(kml_filename, "rb") as file: # Open the file in binary mode coverage_data = file.read() # Read the entire content of the file into memory - if fabric_files is None: + fabric = load_fabric_gdf(folderid) + if fabric is None: raise FileNotFoundError("Fabric or coverage file not found in the database") - fabric_arr = [] - for fabric_file in fabric_files: - tempdf = pandas.read_csv(StringIO(fabric_file.data.decode())) - fabric_arr.append(tempdf) - df = pandas.concat(fabric_arr) - - fabric = geopandas.GeoDataFrame( - df, - crs="EPSG:4326", - geometry=[shapely.geometry.Point(xy) for xy in zip(df.longitude, df.latitude)], - ) - wireless_coverage = read_geo_bytes(coverage_data, ".kml") wireless_coverage = wireless_coverage.to_crs("EPSG:4326") @@ -488,20 +516,14 @@ def preview_wireless_locations(folderid, kml_filename): return bsl_fabric_in_wireless -def compute_wired_locations(folderid, kmlid, download, upload, tech, latency, category, session): - - # Fetch the active (BSL) fabric from the database — never non_bsl / - # supplemental CSVs (see compute_wireless_locations). - fabric_files = get_files_by_type(folderid, "fabric") - if not fabric_files: +def compute_wired_locations( + folderid, kmlid, download, upload, tech, latency, category, session, fabric=None +): + if fabric is None: + fabric = load_fabric_gdf(folderid) + if fabric is None: raise ValueError("No fabric file found") - fabric_arr = [] - for fabric_file in fabric_files: - tempdf = pandas.read_csv(StringIO(fabric_file.data.decode())) - fabric_arr.append(tempdf) - df = pandas.concat(fabric_arr) - # Fetch Fiber file from database fiber_file_record = get_file_with_id(kmlid) if not fiber_file_record: @@ -509,12 +531,6 @@ def compute_wired_locations(folderid, kmlid, download, upload, tech, latency, ca f"No file found with name {fiber_file_record.name} and id {fiber_file_record.id}" ) - fabric = geopandas.GeoDataFrame( - df, - crs="EPSG:4326", - geometry=[shapely.geometry.Point(xy) for xy in zip(df.longitude, df.latitude)], - ) - # Per-file override (files & plans "advanced" setting); NULL keeps the # pipeline's longstanding 100 m default. buffer_meters = fiber_file_record.coverage_buffer_m or 100 @@ -549,14 +565,19 @@ def compute_wired_locations(folderid, kmlid, download, upload, tech, latency, ca return res -def add_network_data(folderid, kmlid, download, upload, tech, type, latency, category, session): +def add_network_data( + folderid, kmlid, download, upload, tech, type, latency, category, session, fabric=None +): + """Compute one coverage file's served locations. `fabric` is the optional + preloaded load_fabric_gdf() frame — pass it when computing several files + so the fabric is parsed once, not per file.""" res = False if type == 0: res = compute_wired_locations( - folderid, kmlid, download, upload, tech, latency, category, session + folderid, kmlid, download, upload, tech, latency, category, session, fabric=fabric ) elif type == 1: res = compute_wireless_locations( - folderid, kmlid, download, upload, tech, latency, category, session + folderid, kmlid, download, upload, tech, latency, category, session, fabric=fabric ) return res diff --git a/back-end/controllers/database_controller/setting_ops.py b/back-end/controllers/database_controller/setting_ops.py index dba2ff4..57eefd9 100644 --- a/back-end/controllers/database_controller/setting_ops.py +++ b/back-end/controllers/database_controller/setting_ops.py @@ -21,3 +21,10 @@ def set_setting(key, value, session): def get_site_theme(session): theme = get_setting("site_theme", DEFAULT_THEME, session) return theme if theme in SITE_THEMES else DEFAULT_THEME + + +def export_max_service_only(session): + """When on, the BDC export reports only the fastest claim per location + instead of every (location, technology) claim. Default OFF — the BDC + accepts multiple technology claims per location.""" + return get_setting("export_max_service_only", "0", session) == "1" diff --git a/back-end/controllers/database_controller/vt_ops.py b/back-end/controllers/database_controller/vt_ops.py index fcb6414..d96e6c7 100755 --- a/back-end/controllers/database_controller/vt_ops.py +++ b/back-end/controllers/database_controller/vt_ops.py @@ -1,4 +1,4 @@ -import json +import math import os import sqlite3 import subprocess @@ -6,6 +6,7 @@ from datetime import datetime from multiprocessing import Lock +import orjson import psycopg2 from psycopg2 import Binary from psycopg2.extras import execute_values @@ -24,6 +25,25 @@ db_lock = Lock() +# The full pyramid is built in TWO tippecanoe runs merged into one tileset: +# +# z0-8 (overview) — density dropping allowed: a z0 tile contains every +# fabric point, so thinning is mandatory there. These zooms are only +# ever produced by full rebuilds. +# z9-16 (detail) — built with NO dropping of any kind (-pk -pf, no +# --drop-densest-as-needed, no byte cap) so a tile's bytes are a pure +# function of the features that intersect it. tippecanoe's +# --drop-densest-as-needed shares the min-gap it discovers on an +# oversized tile across the whole zoom level, making tile content +# depend on OTHER tiles' density — which both silently thinned dense +# areas and would make regenerating just an edited region produce +# different bytes than a full rebuild. Purity at z9-16 is what lets +# an edit retile regenerate only its dirty region and splice the rows +# into the live tileset. +TIPPECANOE_SHARED = "--base-zoom=7 -P --force --use-attribute-for-id=location_id --layer=data" +TIPPECANOE_LOW = f"-z 8 --maximum-tile-bytes=3000000 --drop-densest-as-needed {TIPPECANOE_SHARED}" +TIPPECANOE_HIGH = f"-Z 9 -z 16 -pk -pf {TIPPECANOE_SHARED}" + def _features_from_gdf(gdf, name, skip_points=False, keep_types=None): """Build GeoJSON Feature dicts from a GeoDataFrame, mirroring the legacy @@ -73,66 +93,61 @@ def read_geojson(fileid, session): ) -def add_values_to_VT(geojson_file_path, mbtiles_file_path, folderid): - with sqlite3.connect(mbtiles_file_path) as mb_conn: - mb_c = mb_conn.cursor() - mb_c.execute( - """ - SELECT zoom_level, tile_column, tile_row, tile_data - FROM tiles - """ - ) - - # Create a new connection to Postgres - conn = psycopg2.connect(DATABASE_URL) - cur = conn.cursor() - - try: - with open(mbtiles_file_path, "rb") as file: - mbtiles_data = Binary(file.read()) +def add_values_to_VT(geojson_file_path, mbtiles_file_paths, folderid): + """Store the tiles from one or more freshly built .mbtiles files as ONE + tileset: a single mbtiles anchor row (no file blob — vector_tiles rows + are the only thing ever served) plus all the tile rows. The zoom ranges + of the input files must be disjoint (z0-8 + z9-16).""" + conn = psycopg2.connect(DATABASE_URL) + cur = conn.cursor() - cur.execute("SELECT COUNT(*) FROM mbtiles WHERE folder_id = %s", (folderid,)) - count = cur.fetchone()[0] - cur.execute('SELECT "name" FROM "folder" WHERE id = %s', (folderid,)) - foldername = cur.fetchone()[0] - new_filename = f"{foldername}-{count + 1}.mbtiles" - - cur.execute( - """ - INSERT INTO mbtiles (tile_data, filename, timestamp, folder_id) - VALUES (%s, %s, %s, %s) RETURNING id - """, - (mbtiles_data, new_filename, datetime.now(), folderid), - ) + try: + cur.execute("SELECT COUNT(*) FROM mbtiles WHERE folder_id = %s", (folderid,)) + count = cur.fetchone()[0] + cur.execute('SELECT "name" FROM "folder" WHERE id = %s', (folderid,)) + foldername = cur.fetchone()[0] + new_filename = f"{foldername}-{count + 1}.mbtiles" - mbt_id = cur.fetchone()[0] + cur.execute( + """ + INSERT INTO mbtiles (tile_data, filename, timestamp, folder_id) + VALUES (%s, %s, %s, %s) RETURNING id + """, + (None, new_filename, datetime.now(), folderid), + ) - data = [(row[0], row[1], row[2], Binary(row[3]), mbt_id) for row in mb_c] + mbt_id = cur.fetchone()[0] + for mbtiles_file_path in mbtiles_file_paths: + with sqlite3.connect(mbtiles_file_path) as mb_conn: + mb_c = mb_conn.cursor() + mb_c.execute("SELECT zoom_level, tile_column, tile_row, tile_data FROM tiles") + data = [(row[0], row[1], row[2], Binary(row[3]), mbt_id) for row in mb_c] execute_values( cur, """ - INSERT INTO vector_tiles (zoom_level, tile_column, tile_row, tile_data, mbtiles_id) + INSERT INTO vector_tiles (zoom_level, tile_column, tile_row, tile_data, mbtiles_id) VALUES %s """, data, ) - # Commit the transaction - conn.commit() - except psycopg2.Error as e: - print(f"Database error occurred: {e}") - conn.rollback() - return -1 - except Exception as e: - print(f"Unexpected error occurred: {e}") - conn.rollback() - return -1 - finally: - cur.close() - conn.close() - os.remove(mbtiles_file_path) - os.remove(geojson_file_path) + conn.commit() + except psycopg2.Error as e: + print(f"Database error occurred: {e}") + conn.rollback() + return -1 + except Exception as e: + print(f"Unexpected error occurred: {e}") + conn.rollback() + return -1 + finally: + cur.close() + conn.close() + for mbtiles_file_path in mbtiles_file_paths: + if os.path.exists(mbtiles_file_path): + os.remove(mbtiles_file_path) + os.remove(geojson_file_path) return 1 @@ -187,56 +202,243 @@ def add_values_to_VT(geojson_file_path, mbtiles_file_path, folderid): # return None -def run_tippecanoe(command, folderid, geojsonpath, mbtilepath): +def run_tippecanoe(command): result = subprocess.run(command, shell=True, check=True, stderr=subprocess.PIPE) if result.stderr: print("Tippecanoe stderr:", result.stderr.decode()) - add_values_to_VT(geojsonpath, mbtilepath, folderid) return result.returncode +def _point_features(network_data): + """The point half of the tile feature stream, in get_kml_data order. Both + the full build and the splice MUST build features this way (same shape, + same relative order) for spliced tiles to byte-match a full rebuild.""" + return [ + { + "type": "Feature", + "properties": { + "location_id": point["location_id"], + "served": point["served"], + "address": point["address"], + "wireless": point["wireless"], + "lte": point["lte"], + "network_coverages": point["coveredLocations"], + "maxDownloadNetwork": point["maxDownloadNetwork"], + "maxDownloadSpeed": point["maxDownloadSpeed"], + "bsl": point["bsl"], + "feature_type": "Point", + }, + "geometry": { + "type": "Point", + "coordinates": [point["longitude"], point["latitude"]], + }, + } + for point in network_data + ] + + +def _write_ldjson(path, features): + # Newline-delimited features (not one giant FeatureCollection): orjson is + # much faster than json for the big point sets, and tippecanoe's -P can + # only parallelize input parsing on line-delimited input. + with open(path, "wb") as f: + for feat in features: + f.write(orjson.dumps(feat)) + f.write(b"\n") + + def create_tiles(geojson_array, folderid, session): network_data = get_kml_data(folderid, session) if network_data: - point_geojson = { - "type": "FeatureCollection", - "features": [ - { - "type": "Feature", - "properties": { - "location_id": point["location_id"], - "served": point["served"], - "address": point["address"], - "wireless": point["wireless"], - "lte": point["lte"], - "network_coverages": point["coveredLocations"], - "maxDownloadNetwork": point["maxDownloadNetwork"], - "maxDownloadSpeed": point["maxDownloadSpeed"], - "bsl": point["bsl"], - "feature_type": "Point", - }, - "geometry": { - "type": "Point", - "coordinates": [point["longitude"], point["latitude"]], - }, - } - for point in network_data - ], - } - - # print(geojson_array) - point_geojson["features"].extend(geojson for geojson in geojson_array) + features = _point_features(network_data) + features.extend(geojson_array) uuid_str = str(uuid.uuid4()) unique_geojson_filename = f"data{uuid_str}.geojson" + _write_ldjson(unique_geojson_filename, features) + + low_file = f"output{uuid_str}-low.mbtiles" + high_file = f"output{uuid_str}-high.mbtiles" + run_tippecanoe(f"tippecanoe -o {low_file} {TIPPECANOE_LOW} {unique_geojson_filename}") + run_tippecanoe(f"tippecanoe -o {high_file} {TIPPECANOE_HIGH} {unique_geojson_filename}") + add_values_to_VT(unique_geojson_filename, [low_file, high_file], folderid) + + +# ---- splice: regenerate only the z9-16 tiles a change touched -------------- + +SPLICE_MIN_Z = 9 +MAX_Z = 16 +# tippecanoe renders features up to its --buffer (default 5/256 of a tile) +# beyond a tile's edge; 16/256 of a z9 tile safely covers that overhang at +# every spliced zoom. +TILE_BUFFER_MARGIN = 16 / 256 + + +def _lonlat_to_z9_tile_f(lon, lat): + n = 2**SPLICE_MIN_Z + x = (lon + 180) / 360 * n + r = math.radians(lat) + y = (1 - math.log(math.tan(r) + 1 / math.cos(r)) / math.pi) / 2 * n + return x, y + + +def _z9_tile_f_to_lonlat(xf, yf): + n = 2**SPLICE_MIN_Z + lon = xf / n * 360 - 180 + lat = math.degrees(math.atan(math.sinh(math.pi * (1 - 2 * yf / n)))) + return lon, lat + + +def dirty_z9_ranges(bboxes): + """Snap lon/lat bboxes ([minx, miny, maxx, maxy]) to inclusive z9 XYZ + tile ranges (x0, x1, y0, y1). Each bbox is expanded by the tile-buffer + overhang BEFORE snapping: a changed point within buffer distance of a + region edge is also rendered by the neighboring tile, so that neighbor + must be regenerated too.""" + n = 2**SPLICE_MIN_Z + ranges = [] + for minx, miny, maxx, maxy in bboxes: + x0f, y0f = _lonlat_to_z9_tile_f(minx, maxy) # NW corner + x1f, y1f = _lonlat_to_z9_tile_f(maxx, miny) # SE corner + ranges.append( + ( + max(0, math.floor(x0f - TILE_BUFFER_MARGIN)), + min(n - 1, math.floor(x1f + TILE_BUFFER_MARGIN)), + max(0, math.floor(y0f - TILE_BUFFER_MARGIN)), + min(n - 1, math.floor(y1f + TILE_BUFFER_MARGIN)), + ) + ) + return ranges - with open(unique_geojson_filename, "w") as f: - json.dump(point_geojson, f) - outputFile = f"output{uuid_str}.mbtiles" - command = f"tippecanoe -o {outputFile} --base-zoom=7 -P --maximum-tile-bytes=3000000 -z 16 --drop-densest-as-needed {unique_geojson_filename} --force --use-attribute-for-id=location_id --layer=data" - run_tippecanoe(command, folderid, unique_geojson_filename, outputFile) +def _selection_box(rng): + """The lon/lat box of features a splice run needs for this region: the + region itself plus the buffer margin, so region-edge tiles get their full + buffer content.""" + x0, x1, y0, y1 = rng + minx, maxy = _z9_tile_f_to_lonlat(x0 - TILE_BUFFER_MARGIN, y0 - TILE_BUFFER_MARGIN) + maxx, miny = _z9_tile_f_to_lonlat(x1 + 1 + TILE_BUFFER_MARGIN, y1 + 1 + TILE_BUFFER_MARGIN) + return minx, miny, maxx, maxy + + +def _in_ranges(z, x, y_xyz, ranges): + """Is the (XYZ) tile inside any dirty z9 range (by z9 ancestor)?""" + k = z - SPLICE_MIN_Z + ax, ay = x >> k, y_xyz >> k + return any(x0 <= ax <= x1 and y0 <= ay <= y1 for x0, x1, y0, y1 in ranges) + + +def folder_coverage_features(folderid, session): + """Every coverage file's line/polygon features, in deterministic file + order — the geometry half of the tile feature stream.""" + from .file_ops import get_files_with_postfix + + features = [] + for kml_f in get_files_with_postfix(folderid, ".kml", session): + features.extend(read_kml(kml_f.id, session)) + for geojson_f in get_files_with_postfix(folderid, ".geojson", session): + features.extend(read_geojson(geojson_f.id, session)) + return features + + +def splice_tiles(folderid, bboxes, session): + """Regenerate the z9-16 tiles in the regions the given lon/lat bboxes + touch and replace those rows INSIDE the folder's current tileset (the + serving path never changes; z0-8 overview tiles go deliberately stale + until a settle rebuild). Because z9-16 is built drop-free, the spliced + tiles are byte-identical to what a full rebuild would produce. + + The splice input is the points inside the region + margin and every + coverage geometry passed WHOLE — clipping geometries (even with + tippecanoe's own --clip-bounding-box) perturbs polygon simplification + inside the region; out-of-region output tiles are simply discarded. + + Returns True on success, False when there's nothing to splice into or + nothing to draw (callers fall back to a full rebuild).""" + current = ( + session.query(mbtiles) + .filter(mbtiles.folder_id == folderid) + .order_by(desc(mbtiles.timestamp)) + .first() + ) + if current is None or not bboxes: + return False + network_data = get_kml_data(folderid, session) + if not network_data: + return False + + ranges = dirty_z9_ranges(bboxes) + boxes = [_selection_box(r) for r in ranges] + points = [ + f + for f in _point_features(network_data) + if any( + minx <= f["geometry"]["coordinates"][0] <= maxx + and miny <= f["geometry"]["coordinates"][1] <= maxy + for minx, miny, maxx, maxy in boxes + ) + ] + features = points + folder_coverage_features(folderid, session) + + uuid_str = str(uuid.uuid4()) + src = f"splice{uuid_str}.geojson" + out = f"splice{uuid_str}.mbtiles" + _write_ldjson(src, features) + try: + run_tippecanoe(f"tippecanoe -o {out} {TIPPECANOE_HIGH} {src}") + with sqlite3.connect(out) as mb_conn: + rows = mb_conn.execute( + "SELECT zoom_level, tile_column, tile_row, tile_data FROM tiles" + ).fetchall() + finally: + for path in (src, out): + if os.path.exists(path): + os.remove(path) + + new_tiles = [ + (z, x, y_tms, Binary(bytes(data)), current.id) + for z, x, y_tms, data in rows + if _in_ranges(z, x, (2**z - 1) - y_tms, ranges) + ] + + # Delete-then-insert the region's rows in ONE transaction, so a viewer + # never sees the region empty mid-splice. + conn = psycopg2.connect(DATABASE_URL) + cur = conn.cursor() + try: + for z in range(SPLICE_MIN_Z, MAX_Z + 1): + k = z - SPLICE_MIN_Z + n = 2**z + for x0, x1, y0, y1 in ranges: + ya, yb = y0 << k, ((y1 + 1) << k) - 1 # XYZ rows + cur.execute( + """ + DELETE FROM vector_tiles + WHERE mbtiles_id = %s AND zoom_level = %s + AND tile_column BETWEEN %s AND %s + AND tile_row BETWEEN %s AND %s + """, + (current.id, z, x0 << k, ((x1 + 1) << k) - 1, n - 1 - yb, n - 1 - ya), + ) + if new_tiles: + execute_values( + cur, + """ + INSERT INTO vector_tiles (zoom_level, tile_column, tile_row, tile_data, mbtiles_id) + VALUES %s + """, + new_tiles, + ) + conn.commit() + except Exception as e: + print(f"Splice failed, tiles unchanged: {e}") + conn.rollback() + return False + finally: + cur.close() + conn.close() + return True def retrieve_tiles(zoom, x, y, folderid): diff --git a/back-end/database/models.py b/back-end/database/models.py index 8fae15f..912e612 100755 --- a/back-end/database/models.py +++ b/back-end/database/models.py @@ -492,9 +492,11 @@ class mbtiles(Base): vector_tiles = relationship("vector_tiles", back_populates="mbtiles", cascade="all, delete") def copy(self, session, new_folder_id): + # The vector_tiles rows are the tileset; the mbtiles file blob is no + # longer stored (nothing ever read it back). new_mbtile = mbtiles( filename=self.filename, - tile_data=self.tile_data, + tile_data=None, timestamp=datetime.now(), folder_id=new_folder_id, ) diff --git a/back-end/pyproject.toml b/back-end/pyproject.toml index edb725d..1bcea8c 100644 --- a/back-end/pyproject.toml +++ b/back-end/pyproject.toml @@ -38,6 +38,7 @@ dependencies = [ "Pillow>=10.3", "shortuuid>=1.0.11", "flask-limiter>=3.8", + "orjson>=3.11.9", ] [dependency-groups] diff --git a/back-end/routes/admin_ui.py b/back-end/routes/admin_ui.py index cd549a4..ec864da 100644 --- a/back-end/routes/admin_ui.py +++ b/back-end/routes/admin_ui.py @@ -175,8 +175,8 @@ def audit(): @bp.route("/admin/settings", methods=["GET", "POST"]) @require_platform_admin def settings(): - """Site-wide settings (currently just the theme every provider-facing page - renders with). Theme classes are defined in static/app/bdk.css.""" + """Site-wide settings: the theme every provider-facing page renders with + (classes in static/app/bdk.css) and the export max-service-only toggle.""" from controllers.database_controller import setting_ops session = get_session() @@ -188,11 +188,18 @@ def settings(): error = "Unknown theme." else: setting_ops.set_setting("site_theme", theme, session) - log_action("set_site_theme", user_id=g.admin_user.id, details={"theme": theme}) + max_service = "1" if request.form.get("export_max_service_only") else "0" + setting_ops.set_setting("export_max_service_only", max_service, session) + log_action( + "set_site_settings", + user_id=g.admin_user.id, + details={"theme": theme, "export_max_service_only": max_service}, + ) saved = True ctx = { "themes": setting_ops.SITE_THEMES, "current": setting_ops.get_site_theme(session), + "max_service_only": setting_ops.export_max_service_only(session), "saved": saved, "error": error, } diff --git a/back-end/templates/admin/settings.html b/back-end/templates/admin/settings.html index f35ca1e..79fa635 100644 --- a/back-end/templates/admin/settings.html +++ b/back-end/templates/admin/settings.html @@ -14,6 +14,13 @@
Applied to every provider-facing page site-wide. Providers don't choose themes.
+ +When on, generated BDC reports include just the fastest claim per location + (by download, then upload) instead of every technology claim. The BDC accepts multiple + claims per location, so this is off by default.
{% endblock %} diff --git a/back-end/tests/test_admin_ui.py b/back-end/tests/test_admin_ui.py index a6ca8ca..aaa998b 100644 --- a/back-end/tests/test_admin_ui.py +++ b/back-end/tests/test_admin_ui.py @@ -190,3 +190,27 @@ def test_settings_rejects_unknown_theme(client): resp = client.post("/admin/settings", data={"csrf_token": csrf, "site_theme": "hotdog-stand"}) assert resp.status_code == 400 assert b"Unknown theme." in resp.data + + +def test_settings_max_service_only_roundtrip(client, db_session): + """The export max-service toggle persists (checkbox on -> '1', absent -> + '0') and is read back by the export-path helper. Default is OFF.""" + from controllers.database_controller.setting_ops import export_max_service_only + + assert export_max_service_only(db_session) is False + + _, csrf = _admin_session(client) + resp = client.post( + "/admin/settings", + data={"csrf_token": csrf, "site_theme": "civic-light", "export_max_service_only": "1"}, + ) + assert resp.status_code == 200 and b"Saved." in resp.data + db_session.expire_all() + assert export_max_service_only(db_session) is True + assert b"checked" in client.get("/admin/settings").data + + # Unchecking (the field absent from the POST) turns it back off. + resp = client.post("/admin/settings", data={"csrf_token": csrf, "site_theme": "civic-light"}) + assert resp.status_code == 200 + db_session.expire_all() + assert export_max_service_only(db_session) is False diff --git a/back-end/tests/test_csv_export.py b/back-end/tests/test_csv_export.py index 8289d8c..54010b4 100644 --- a/back-end/tests/test_csv_export.py +++ b/back-end/tests/test_csv_export.py @@ -47,3 +47,32 @@ def test_provider_and_brand_are_filled(): df = generate_csv_data([_row(1), _row(2)], provider_id=330054, brand_name="Acme") assert (df.provider_id == 330054).all() assert (df.brand_name == "Acme").all() + + +def test_default_reports_every_technology_claim(): + """The BDC accepts multiple technology claims per location; by default a + location under several coverages files a row per technology.""" + rows = [_row(1, tech=50), _row(1, tech=70), _row(1, tech=71)] + df = generate_csv_data(rows, provider_id=330054, brand_name="Acme") + assert set(zip(df.location_id, df.technology)) == {(1, 50), (1, 70), (1, 71)} + + +def test_max_service_only_picks_the_fastest_claim_per_location(): + """With max_service_only, a location files exactly ONE row — the fastest + claim: download desc, then upload desc, then low-latency first, then the + lowest technology code as a deterministic tiebreak. The surviving row + keeps its own values.""" + rows = [ + _row(1, tech=50, dl=1000, ul=1000), + _row(1, tech=70, dl=25, ul=3), # slower -> dropped + _row(2, tech=70, dl=100, ul=50), + _row(2, tech=71, dl=100, ul=75), # same download, faster upload -> wins + _row(3, tech=70, dl=100, ul=20, latency=0), + _row(3, tech=71, dl=100, ul=20, latency=1), # low latency wins the tie + _row(4, tech=71, dl=100, ul=20), + _row(4, tech=70, dl=100, ul=20), # full tie -> lowest tech code + ] + df = generate_csv_data(rows, provider_id=330054, brand_name="Acme", max_service_only=True) + assert set(zip(df.location_id, df.technology)) == {(1, 50), (2, 71), (3, 71), (4, 70)} + kept = df[df.location_id == 1].iloc[0] + assert kept.max_advertised_download_speed == 1000 # the winner's own values diff --git a/back-end/tests/test_prod_golden_csv.py b/back-end/tests/test_prod_golden_csv.py new file mode 100644 index 0000000..a80da32 --- /dev/null +++ b/back-end/tests/test_prod_golden_csv.py @@ -0,0 +1,84 @@ +"""Layer 4b — PROD golden for the EXPORT CSV: replay real production filings +and assert the availability CSV our export path generates matches the bytes +prod actually stored (the BDC report the provider filed). + +This pins the export-layer behavior that kml_data equality cannot see. The +shipped default reports every (location, technology) claim — the BDC accepts +multiple claims per location — which matches the reference era pinned here. +(Some historical snapshots were generated with a since-retired per-location +wireless dedup; they stay in the fixture set as reference only — see +csv-golden.json. The optional export_max_service_only setting, pinned by +unit tests in test_csv_export.py, is the supported variant of that idea.) + +References: prod-extract/