diff --git a/computing/misc/antyodaya/pipeline.py b/computing/misc/antyodaya/pipeline.py index baeace53..3e5aa107 100644 --- a/computing/misc/antyodaya/pipeline.py +++ b/computing/misc/antyodaya/pipeline.py @@ -13,6 +13,7 @@ import pandas as pd from django.conf import settings +from computing.utils import save_layer_info_to_db, update_layer_sync_status from utilities.pipelines import AdminScope, CSAdminSource, StandardRequest, load_config from utilities.pipelines.admin import ( ADMIN_COLUMN_DESCRIPTIONS, @@ -28,7 +29,7 @@ stable_hash, utc_now_text, ) -from utilities.pipelines.publish import publish_gpkg_layer, register_layer +from utilities.pipelines.publish import publish_gpkg_layer from utilities.pipelines.schema import ( STATUS_MATCHED, STATUS_NO_DATA, @@ -597,29 +598,6 @@ def run_antyodaya_pipeline( geoserver = asdict(geoserver_result) geoserver["ok"] = True geoserver["status"] = "published" - if request.publish.register_layers: - result["layer_registration"] = register_layer( - dataset_name=output_config.get("dataset_name", "Antyodaya 2020"), - layer_name=result_name, - scope=request.scope, - workspace=geoserver_workspace, - geoserver_url=geoserver.get("wfs_url"), - algorithm=ALGORITHM, - algorithm_version=ALGORITHM_VERSION, - misc={ - "source_csv": config["sources"]["csv"], - "gpkg_path": result.get("gpkg_path"), - "links_path": result.get("links_path"), - "output_dir": bundle.path.as_posix(), - "geoserver_layer_name": result_name, - "geoserver_url": geoserver.get("wfs_url"), - "rows": result.get("rows"), - "matched_rows": result.get("matched_rows"), - "join_coverage": result.get("join_coverage"), - }, - overwrite=request.publish.overwrite, - ) - result["layer_id"] = (result["layer_registration"] or {}).get("layer_id") except Exception as exc: geoserver = { "ok": False, @@ -631,6 +609,42 @@ def run_antyodaya_pipeline( } timings["publish_geoserver_seconds"] = round(time.perf_counter() - t0, 3) result["geoserver"] = geoserver + if request.publish.register_layers and geoserver and geoserver.get("ok"): + state = request.scope.state_name + district = request.scope.district_name + block = request.scope.tehsil_name + if not (state and district and block): + raise ValueError( + "Layer registration requires state, district, and tehsil names." + ) + dataset_name = output_config.get("dataset_name", "Antyodaya 2020") + layer_id = save_layer_info_to_db( + state=state, + district=district, + block=block, + layer_name=result_name, + asset_id="not applicable: local compute GeoServer layer", + dataset_name=dataset_name, + algorithm=ALGORITHM, + algorithm_version=ALGORITHM_VERSION, + misc={"is_generated_locally": True}, + is_override=request.publish.overwrite, + ) + if layer_id is None: + raise RuntimeError(f"Database registration failed for layer {result_name!r}.") + if update_layer_sync_status( + layer_id=layer_id, sync_to_geoserver=True + ) is None: + raise RuntimeError( + f"GeoServer sync status update failed for layer ID {layer_id}." + ) + result["layer_id"] = layer_id + result["layer_registration"] = { + "ok": True, + "status": "registered", + "dataset": dataset_name, + "layer_id": layer_id, + } if outputs.readme: result["readme_path"] = bundle.write_readme( _readme_lines( diff --git a/computing/misc/facilities/pipeline.py b/computing/misc/facilities/pipeline.py index 413b3fec..7e847551 100644 --- a/computing/misc/facilities/pipeline.py +++ b/computing/misc/facilities/pipeline.py @@ -16,6 +16,7 @@ from django.conf import settings from scipy.spatial import cKDTree +from computing.utils import save_layer_info_to_db, update_layer_sync_status from utilities.pipelines import AdminScope, CSAdminSource, StandardRequest, load_config from utilities.pipelines.admin import ( ADMIN_COLUMN_DESCRIPTIONS, @@ -53,7 +54,6 @@ from utilities.pipelines.publish import ( publish_gpkg_layer, publish_gpkg_layers, - register_layer, ) from utilities.pipelines.unicode import normalize_unicode_frame from nrm_app.celery import app @@ -1134,33 +1134,48 @@ def run_facilities_pipeline( } ).as_posix() if request.publish.register_layers and published_layers: - common_misc = { - "source_facilities_gpkg": config["sources"]["facilities_gpkg"], - "gpkg_path": result.get("gpkg_path"), - "facility_points_gpkg_path": result.get("facility_points_gpkg_path"), - "links_path": result.get("links_path"), - "output_dir": bundle.path.as_posix(), - "village_rows": result.get("village_rows"), - "nearest_rows": result.get("nearest_rows"), - "inventory_rows": result.get("inventory_rows"), - } + state = request.scope.state_name + district = request.scope.district_name + block = request.scope.tehsil_name + if not (state and district and block): + raise ValueError( + "Layer registration requires state, district, and tehsil names." + ) registrations: dict[str, dict[str, Any]] = {} for role, published in published_layers.items(): - registrations[role] = register_layer( - dataset_name=( - output_config.get("dataset_name", "Facilities Proximity") - if role == "village_properties" - else output_config.get("points_dataset_name", "Facilities Points") - ), + dataset_name = ( + output_config.get("dataset_name", "Facilities Proximity") + if role == "village_properties" + else output_config.get("points_dataset_name", "Facilities Points") + ) + layer_id = save_layer_info_to_db( + state=state, + district=district, + block=block, layer_name=published["layer_name"], - scope=request.scope, - workspace=published["workspace"], - geoserver_url=published["wfs_url"], + asset_id="not applicable: local compute GeoServer layer", + dataset_name=dataset_name, algorithm=ALGORITHM, algorithm_version=ALGORITHM_VERSION, - misc={**common_misc, "output_role": role}, - overwrite=request.publish.overwrite, + misc={"is_generated_locally": True}, + is_override=request.publish.overwrite, ) + if layer_id is None: + raise RuntimeError( + f"Database registration failed for layer {published['layer_name']!r}." + ) + if update_layer_sync_status( + layer_id=layer_id, sync_to_geoserver=True + ) is None: + raise RuntimeError( + f"GeoServer sync status update failed for layer ID {layer_id}." + ) + registrations[role] = { + "ok": True, + "status": "registered", + "dataset": dataset_name, + "layer_id": layer_id, + } result["layer_registrations"] = registrations result["layer_registration"] = registrations.get("village_properties") result["layer_id"] = (result.get("layer_registration") or {}).get("layer_id") diff --git a/computing/misc/livestocks/pipeline.py b/computing/misc/livestocks/pipeline.py index d8de86a7..bd2b2d71 100644 --- a/computing/misc/livestocks/pipeline.py +++ b/computing/misc/livestocks/pipeline.py @@ -12,6 +12,7 @@ import pandas as pd from django.conf import settings +from computing.utils import save_layer_info_to_db, update_layer_sync_status from utilities.pipelines import AdminScope, CSAdminSource, StandardRequest, load_config from utilities.pipelines.admin import ( ADMIN_COLUMN_DESCRIPTIONS, @@ -27,7 +28,7 @@ stable_hash, utc_now_text, ) -from utilities.pipelines.publish import publish_gpkg_layer, register_layer +from utilities.pipelines.publish import publish_gpkg_layer from utilities.pipelines.schema import ( STATUS_MATCHED, STATUS_NO_DATA, @@ -486,29 +487,6 @@ def run_livestocks_pipeline( geoserver = asdict(geoserver_result) geoserver["ok"] = True geoserver["status"] = "published" - if request.publish.register_layers: - result["layer_registration"] = register_layer( - dataset_name=output_config.get("dataset_name", "Livestock Census"), - layer_name=layer_name, - scope=request.scope, - workspace=geoserver_workspace, - geoserver_url=geoserver.get("wfs_url"), - algorithm=ALGORITHM, - algorithm_version=ALGORITHM_VERSION, - misc={ - "source_csv": config["sources"]["csv"], - "gpkg_path": result.get("gpkg_path"), - "links_path": result.get("links_path"), - "output_dir": bundle.path.as_posix(), - "geoserver_layer_name": layer_name, - "geoserver_url": geoserver.get("wfs_url"), - "rows": result.get("rows"), - "matched_rows": result.get("matched_rows"), - "join_coverage": result.get("join_coverage"), - }, - overwrite=request.publish.overwrite, - ) - result["layer_id"] = (result["layer_registration"] or {}).get("layer_id") except Exception as exc: geoserver = { "ok": False, @@ -520,6 +498,42 @@ def run_livestocks_pipeline( } timings["publish_geoserver_seconds"] = round(time.perf_counter() - t0, 3) result["geoserver"] = geoserver + if request.publish.register_layers and geoserver and geoserver.get("ok"): + state = request.scope.state_name + district = request.scope.district_name + block = request.scope.tehsil_name + if not (state and district and block): + raise ValueError( + "Layer registration requires state, district, and tehsil names." + ) + dataset_name = output_config.get("dataset_name", "Livestock Census") + layer_id = save_layer_info_to_db( + state=state, + district=district, + block=block, + layer_name=layer_name, + asset_id="not applicable: local compute GeoServer layer", + dataset_name=dataset_name, + algorithm=ALGORITHM, + algorithm_version=ALGORITHM_VERSION, + misc={"is_generated_locally": True}, + is_override=request.publish.overwrite, + ) + if layer_id is None: + raise RuntimeError(f"Database registration failed for layer {layer_name!r}.") + if update_layer_sync_status( + layer_id=layer_id, sync_to_geoserver=True + ) is None: + raise RuntimeError( + f"GeoServer sync status update failed for layer ID {layer_id}." + ) + result["layer_id"] = layer_id + result["layer_registration"] = { + "ok": True, + "status": "registered", + "dataset": dataset_name, + "layer_id": layer_id, + } if outputs.readme: result["readme_path"] = bundle.write_readme( _readme_lines( diff --git a/utilities/pipelines/__init__.py b/utilities/pipelines/__init__.py index a0960d1e..1bc80a6a 100644 --- a/utilities/pipelines/__init__.py +++ b/utilities/pipelines/__init__.py @@ -7,7 +7,6 @@ """ from .admin import AdminScope, CSAdminSource -from .publish import register_layer from .schema import StandardRequest, api_request_payload, load_config __all__ = [ @@ -15,6 +14,5 @@ "CSAdminSource", "StandardRequest", "api_request_payload", - "register_layer", "load_config", ] diff --git a/utilities/pipelines/publish.py b/utilities/pipelines/publish.py index ea8f6a6b..f697a5af 100644 --- a/utilities/pipelines/publish.py +++ b/utilities/pipelines/publish.py @@ -405,64 +405,3 @@ def publish_gpkg_layers( property_count=len(verification["properties"]), ) return results - - -def register_layer( - *, - dataset_name: str, - layer_name: str, - scope: Any, - workspace: str, - geoserver_url: str, - algorithm: str, - algorithm_version: str, - misc: dict[str, Any] | None = None, - overwrite: bool = False, -) -> dict[str, Any]: - """Register a published layer in the Core Stack layer database. - - Uses the same `save_layer_info_to_db` path as the GEE-backed pipelines, so - local pipeline layers appear alongside every other layer. The `Layer` model - is keyed on state/district/block, so only tehsil-scoped runs can register; - other scopes are reported as skipped rather than failing the run. - - Never raises: the run has already succeeded by the time this is called, so - a database problem is reported in the result instead of losing the outputs. - """ - - level = str(getattr(scope, "level", "") or "").lower() - if level not in {"tehsil", "block"}: - return {"ok": False, "status": "skipped", "reason": f"layer registration needs a tehsil scope, got {level!r}"} - state, district, block = scope.state_name, scope.district_name, scope.tehsil_name - if not (state and district and block): - return {"ok": False, "status": "skipped", "reason": "scope is missing state, district, or tehsil name"} - - try: - from computing.models import Dataset, LayerType - from computing.utils import save_layer_info_to_db, update_layer_sync_status - - Dataset.objects.get_or_create( - name=dataset_name, - defaults={"layer_type": LayerType.VECTOR, "workspace": workspace}, - ) - layer_id = save_layer_info_to_db( - state=state, - district=district, - block=block, - layer_name=layer_name, - asset_id=geoserver_url, - dataset_name=dataset_name, - algorithm=algorithm, - algorithm_version=algorithm_version, - misc={"is_generated_locally": True, "geoserver_workspace": workspace, **(misc or {})}, - is_override=overwrite, - is_gee_asset=False, - ) - if not layer_id: - return {"ok": False, "status": "not_registered", "dataset": dataset_name, - "reason": "save_layer_info_to_db returned no layer id (check state/district/block exist in the SOI tables)"} - update_layer_sync_status(layer_id=layer_id, sync_to_geoserver=True) - return {"ok": True, "status": "registered", "dataset": dataset_name, "layer_id": layer_id} - except Exception as exc: - return {"ok": False, "status": "registration_failed", "dataset": dataset_name, - "error_type": exc.__class__.__name__, "error": str(exc)[:500]} diff --git a/utilities/pipelines/tabular.py b/utilities/pipelines/tabular.py index 29fb3976..f3aad2c4 100644 --- a/utilities/pipelines/tabular.py +++ b/utilities/pipelines/tabular.py @@ -135,7 +135,12 @@ def fetch_by_values( values = [value for value in values if value is not None and value == value] if not values: - return pd.DataFrame(columns=list(columns or [])) + empty_columns = ( + columns + if columns is not None + else (*self.key_columns, *(self.source_columns or ())) + ) + return pd.DataFrame(columns=list(dict.fromkeys(empty_columns))) if not self.is_fresh(): self.materialize() select_columns = "*" if columns is None else ", ".join(quote_identifier(col) for col in columns)