diff --git a/computing/api.py b/computing/api.py index de3af6bc..543e4a6c 100644 --- a/computing/api.py +++ b/computing/api.py @@ -1513,25 +1513,71 @@ def generate_ndvi_timeseries(request): def generate_zoi_to_gee(request): print("Inside generate zoi layers") try: - state = request.data.get("state").lower() - district = request.data.get("district").lower() - block = request.data.get("block").lower() + state = request.data.get("state") + district = request.data.get("district") + block = request.data.get("block") gee_account_id = request.data.get("gee_account_id") + start_date = request.data.get("start_date") or request.data.get("startDate") + end_date = request.data.get("end_date") or request.data.get("endDate") + start_year = request.data.get("start_year") or request.data.get("startYear") + end_year = request.data.get("end_year") or request.data.get("endYear") + + if not state or not district or not block: + return Response( + {"error": "state, district, and block are required."}, + status=status.HTTP_400_BAD_REQUEST, + ) + + # Allow start_year/end_year as hydrological-year shorthand. + if not start_date and start_year is not None: + start_date = f"{int(start_year)}-07-01" + if not end_date and end_year is not None: + end_date = f"{int(end_year) + 1}-06-30" + + if not start_date or not end_date: + return Response( + { + "error": ( + "start_date and end_date are required (YYYY-MM-DD), " + "or provide start_year and end_year (hydrological years)." + ) + }, + status=status.HTTP_400_BAD_REQUEST, + ) + + from computing.zoi_layers.zoi import _resolve_zoi_time_window + + try: + start_date, end_date, _, _ = _resolve_zoi_time_window(start_date, end_date) + except ValueError as exc: + return Response({"error": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + + state = state.lower() + district = district.lower() + block = block.lower() + generate_zoi.apply_async( kwargs={ "state": state, "district": district, "block": block, "gee_account_id": gee_account_id, + "start_date": start_date, + "end_date": end_date, }, queue="waterbody", ) return Response( - {"Success": "Successfully initiated"}, status=status.HTTP_200_OK + { + "Success": "Successfully initiated", + "start_date": start_date, + "end_date": end_date, + }, + status=status.HTTP_200_OK, ) except Exception as e: - print("Exception in generate_mining_to_gee api :: ", e) + print("Exception in generate_zoi_to_gee api :: ", e) return Response({"Exception": e}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) diff --git a/computing/utils.py b/computing/utils.py index 1e4732e2..2faf8919 100644 --- a/computing/utils.py +++ b/computing/utils.py @@ -1,1191 +1,1198 @@ -import copy -import json -import logging -import os -import shutil -import zipfile -from datetime import datetime, timedelta - -import ee -import fiona -import geopandas as gpd -import requests -from django.conf import settings -from shapely.geometry import shape -from shapely.validation import explain_validity - -from computing.models import Dataset, Layer -from geoadmin.models import ( - DistrictSOI, - State_Disritct_Block_Properties, - StateSOI, - TehsilSOI, -) -from projects.models import Project -from utilities.constants import ( - ADMIN_BOUNDARY_OUTPUT_DIR, - GEE_ASSET_PATH, - GEE_HELPER_PATH, - GEE_PATHS, - SHAPEFILE_DIR, -) -from utilities.gee_utils import ( - check_task_status, - ee_initialize, - get_gee_asset_path, - get_gee_dir_path, - get_geojson_from_gcs, - is_asset_public, - is_gee_asset_exists, - sync_vector_to_gcs, - valid_gee_text, -) -from utilities.geoserver_utils import Geoserver -from django.core.mail import EmailMessage, get_connection -import time - - -logger = logging.getLogger(__name__) - - -def generate_shape_files(path): - gdf = gpd.read_file(path + ".json") - if os.path.exists(path): - # Only replace the target shapefile directory. Removing the parent - # state/workspace directory here corrupts sibling outputs on reruns. - shutil.rmtree(path) - - os.makedirs(os.path.dirname(path), exist_ok=True) - gdf.to_file( - path, - driver="ESRI Shapefile", - ) - return path - - -def convert_to_zip(dir_name, file_type): - if file_type == "gpkg": - with zipfile.ZipFile(dir_name + ".zip", "w", zipfile.ZIP_DEFLATED) as zipf: - zipf.write(dir_name + ".gpkg", arcname=os.path.basename(dir_name + ".gpkg")) - return dir_name + ".zip" - else: - return shutil.make_archive(dir_name, "zip", dir_name + "/") - - -def push_shape_to_geoserver( - path, store_name=None, workspace=None, layer_name=None, file_type="shp" -): - geo = Geoserver() - - print(f"layer_name: {layer_name}") - if layer_name: - try: - print(f"Attempting to delete store: {layer_name}") - geo.delete_vector_store(workspace=workspace, store=layer_name) - print(f"Successfully deleted store: {layer_name}") - except Exception as e: - print(f"Store does not exist or error deleting: {str(e)}") - - zip_path = convert_to_zip(path, file_type) - print(f"Zip path: {zip_path}") - print(f"Store name: {store_name}") - print(f"Workspace: {workspace}") - - response = geo.create_shp_datastore( - path=zip_path, - store_name=store_name, - workspace=workspace, - file_extension=file_type, - ) - print(f"Response: {response}") - return response - - -def kml_to_geojson(state_name, district_name, block_name, kml_path): - fiona.drvsupport.supported_drivers["kml"] = ( - "rw" # enable KML support which is disabled by default - ) - fiona.drvsupport.supported_drivers["KML"] = ( - "rw" # enable KML support which is disabled by default - ) - gdf = gpd.read_file(kml_path) - geometry_types = gdf.geometry.geometry.type.unique() - state_dir = os.path.join(ADMIN_BOUNDARY_OUTPUT_DIR, state_name) - - for gtype in geometry_types: - df = gdf.loc[gdf.geometry.geometry.type == gtype] - path = os.path.join(state_dir, f"{district_name}_{block_name}_{gtype}") - df.to_file(path + ".json", driver="GeoJSON") - generate_shape_files(path) - push_shape_to_geoserver(path, workspace="test_workspace") - - -def convert_kml_to_shapefile(kml_path, output_dir, shapefile_name): - if not os.path.exists(output_dir + "/" + shapefile_name): - os.makedirs(output_dir + "/" + shapefile_name) - - shapefile_path = os.path.join( - output_dir + "/" + shapefile_name, shapefile_name + ".shp" - ) - print("path path", shapefile_path) - cmd = f"ogr2ogr -f 'ESRI Shapefile' {shapefile_path} {kml_path}" # output.shp input.kml - os.system(command=cmd) - - return output_dir + "/" + shapefile_name - - -def kml_to_shp(state_name, district_name, block_name, kml_path): - shapefile_name = f"{district_name}_{block_name}" - shapefile_layer_path = convert_kml_to_shapefile( - kml_path, SHAPEFILE_DIR, shapefile_name - ) - - push_shape_to_geoserver(shapefile_layer_path, workspace="customkml") - - # os.remove(kml_path) - # shutil.rmtree(shapefile_layer_path) - os.remove(shapefile_layer_path + ".zip") - - -def sync_layer_to_geoserver(state_name, fc, layer_name, workspace): - state_dir = os.path.join("data/fc_to_shape", state_name) - if not os.path.exists(state_dir): - os.mkdir(state_dir) - path = os.path.join(state_dir, f"{layer_name}") - # Write the feature collection into json file - with open(path + ".json", "w") as f: - try: - f.write(f"{json.dumps(fc)}") - except Exception as e: - print(e) - - path = generate_shape_files(path) - return push_shape_to_geoserver(path, workspace=workspace, layer_name=layer_name) - - -def sync_fc_to_geoserver(fc, shp_folder, layer_name, workspace, style_name=None): - try: - geojson_fc = fc.getInfo() - except Exception as e: - print("Exception in getInfo()", e) - task_id = sync_vector_to_gcs(fc, layer_name, "GeoJSON") - check_task_status([task_id]) - - geojson_fc = get_geojson_from_gcs(layer_name) - geo = Geoserver() - if len(geojson_fc["features"]) > 0: - state_dir = os.path.join("data/fc_to_shape", shp_folder) - if not os.path.exists(state_dir): - os.mkdir(state_dir) - path = os.path.join(state_dir, f"{layer_name}") - - # Convert to GeoDataFrame - gdf = gpd.GeoDataFrame.from_features(geojson_fc["features"]) - - # Set CRS (Earth Engine uses EPSG:4326 by default) - gdf.crs = "EPSG:4326" - - gdf = fix_invalid_geometry_in_gdf(gdf) - - # Save as GeoPackage - gdf.to_file(path + ".gpkg", driver="GPKG") - res = push_shape_to_geoserver(path, workspace=workspace, file_type="gpkg") - if style_name: - style_res = geo.publish_style( - layer_name=layer_name, style_name=style_name, workspace=workspace - ) - print("Style response:", style_res) - return res - else: - return "No features in FeatureCollection" - - -def sync_project_fc_to_geoserver(fc, project_name, layer_name, workspace): - print("inside") - print(layer_name) - try: - geojson_fc = fc.getInfo() - except Exception as e: - print("Exception in getInfo()", e) - task_id = sync_vector_to_gcs(fc, layer_name, "GeoJSON") - check_task_status([task_id]) - - geojson_fc = get_geojson_from_gcs(layer_name) - print(len(geojson_fc["features"])) - if len(geojson_fc["features"]) > 0: - state_dir = os.path.join("data/fc_to_shape", project_name) - if not os.path.exists(state_dir): - os.mkdir(state_dir) - path = os.path.join(state_dir, f"{layer_name}") - - # Convert to GeoDataFrame - gdf = gpd.GeoDataFrame.from_features(geojson_fc["features"]) - - # Set CRS (Earth Engine uses EPSG:4326 by default) - gdf.crs = "EPSG:4326" - - gdf = fix_invalid_geometry_in_gdf(gdf) - - # Save as GeoPackage - gdf.to_file(path + ".gpkg", driver="GPKG") - print("pushed to geoserver") - return push_shape_to_geoserver( - path, workspace=workspace, layer_name=layer_name, file_type="gpkg" - ) - else: - print("no features found") - return - - -def to_camelcase(text): - words = text.split() - camelcase = words[0].lower() - for word in words[1:]: - camelcase += word.capitalize() - return camelcase - - -def create_chunk(aoi, description, chunk_size): - size = aoi.size().getInfo() - parts = size // chunk_size - # task_ids = [] - rois = [] - descs = [] - for part in range(parts + 1): - start = part * chunk_size - end = start + chunk_size - block_name_for_parts = description + "_" + str(start) + "-" + str(end) - roi = ee.FeatureCollection(aoi.toList(aoi.size()).slice(start, end)) - if roi.size().getInfo() > 0: - descs.append(block_name_for_parts) - rois.append(roi) - - return rois, descs - - -def merge_chunks( - aoi, - folder_list, - description, - chunk_size, - chunk_asset_path=GEE_HELPER_PATH, - merge_asset_path=GEE_ASSET_PATH, - merge_asset_id=None, -): - print("Merge Chunk task initiated") - ee_initialize() - size = aoi.size().getInfo() - parts = size // chunk_size - assets = [] - for part in range(parts + 1): - start = part * chunk_size - end = start + chunk_size - block_name_for_parts = description + "_" + str(start) + "-" + str(end) - src_asset_id = ( - get_gee_dir_path(folder_list, chunk_asset_path) + block_name_for_parts - ) - if is_gee_asset_exists(src_asset_id): - assets.append(ee.FeatureCollection(src_asset_id)) - - asset = ee.FeatureCollection(assets).flatten() - - asset_id = merge_asset_id or ( - get_gee_dir_path(folder_list, merge_asset_path) + description - ) - try: - # Export an ee.FeatureCollection as an Earth Engine asset. - task = ee.batch.Export.table.toAsset( - **{ - "collection": asset, - "description": description, - "assetId": asset_id, - } - ) - - task.start() - print("Successfully started the merge chunk", task.status()) - return task.status()["id"] - except Exception as e: - print(f"Error occurred in running merge task: {e}") - return None - - -def fix_invalid_geometry_in_gdf(gdf): - invalid = gdf[~gdf.is_valid] - if not invalid.empty: - print("Invalid geometries found:") - for idx, geom in invalid.geometry.items(): - print(f"Index {idx}: {explain_validity(geom)}") - gdf.loc[idx, "geometry"] = gdf.loc[idx, "geometry"].buffer(0) - - return gdf - - -def get_season_key(date): - """Return season key like 'rabi_2017-2018' based on Indian cropping seasons.""" - month = date.month - year = date.year - next_year = year + 1 - - if month in [1, 2]: - return f"rabi_{year - 1}-{year}" # Jan–Feb → Rabi of previous year - elif month in [11, 12]: - return f"rabi_{year}-{next_year}" # Nov–Dec → Rabi starting this year - elif month in [3, 4, 5, 6]: - return f"zaid_{year}-{next_year}" - elif month in [7, 8, 9, 10]: - return f"kharif_{year}-{next_year}" - else: - return None - - -def get_agri_year_key(season_key): - """Convert a season key to agricultural year key (e.g., rabi_2017-2018 → 2017-2018).""" - season, years = season_key.split("_") - start_year, end_year = map(int, years.split("-")) - - if season in ["kharif", "rabi"]: - return f"{start_year}-{end_year}" - elif season == "zaid": - return f"{start_year - 1}-{start_year}" # Zaid 2018-2019 → Agri year 2017-2018 - else: - return None - - -def calculate_precipitation_season( - geojson_filepath, draught_asset_id, start_year=2017, end_year=2024 -): - # Load the GeoJSON file - with open(geojson_filepath, "r") as f: - feature_collection = json.load(f) - - features_ee = [] - - for feature in feature_collection["features"]: - original_props = feature["properties"] - new_props = {} - - # Copy UID - if "uid" in original_props: - new_props["uid"] = original_props["uid"] - - agri_year_totals = {} - - # Parse precipitation date keys - for key, val in original_props.items(): - try: - date = datetime.strptime(key, "%Y-%m-%d") - season_key = get_season_key(date) - if not season_key: - continue - - agri_key = get_agri_year_key(season_key) - if not agri_key: - continue - - agri_start = int(agri_key.split("-")[0]) - if not (start_year <= agri_start <= end_year): - continue - - season = season_key.split("_")[0] # kharif, rabi etc - full_key = f"{season}_{agri_key}" - - agri_year_totals[full_key] = agri_year_totals.get(full_key, 0) + float( - val - ) - - except Exception: - continue - - # Add all seasonal totals to new_props - for agri_key, total in agri_year_totals.items(): - new_props[f"precipitation_{agri_key}"] = total - - # Create EE Feature - geom_ee = ee.Geometry(feature["geometry"]) - feature_ee = ee.Feature(geom_ee, new_props) - features_ee.append(feature_ee) - - # Left side FC - mws_fc = ee.FeatureCollection(features_ee) - - return mws_fc - - -def generate_geojson_with_ci_and_ndvi(zoi_asset, ci_asset, ndvi_asset, proj_id): - # Load project - proj_obj = Project.objects.get(pk=proj_id) - - # Build CI and NDVI asset paths - asset_path_ci = ( - get_gee_dir_path( - [proj_obj.name], asset_path=GEE_PATHS["WATER_REJ"]["GEE_ASSET_PATH"] - ) - + ci_asset - ) - - asset_path_ndvi = ( - get_gee_dir_path( - [proj_obj.name], asset_path=GEE_PATHS["WATER_REJ"]["GEE_ASSET_PATH"] - ) - + ndvi_asset - ) - - # Load FeatureCollections - zoi = ee.FeatureCollection(zoi_asset) - ci = ee.FeatureCollection(asset_path_ci) - ndvi = ee.FeatureCollection(asset_path_ndvi) - - # ------------------------- - # STEP 1: Join ZOI with Cropping Intensity - # ------------------------- - join = ee.Join.inner() - filter = ee.Filter.intersects(leftField=".geo", rightField=".geo") - zoi_ci_joined = join.apply(zoi, ci, filter) - - def merge_zoi_ci(pair): - zoi_feat = ee.Feature(pair.get("primary")) - ci_feat = ee.Feature(pair.get("secondary")) - merged_props = zoi_feat.toDictionary().combine(ci_feat.toDictionary(), True) - return ee.Feature(zoi_feat.geometry(), merged_props) - - zoi_with_ci = ee.FeatureCollection(zoi_ci_joined.map(merge_zoi_ci)) - - # ------------------------- - # STEP 2: Join ZOI+CI with NDVI - # ------------------------- - zoi_ndvi_joined = join.apply(zoi_with_ci, ndvi, filter) - - def merge_zoi_ci_ndvi(pair): - ci_feat = ee.Feature(pair.get("primary")) - ndvi_feat = ee.Feature(pair.get("secondary")) - merged_props = ci_feat.toDictionary().combine(ndvi_feat.toDictionary(), True) - return ee.Feature(ci_feat.geometry(), merged_props) - - final_merged = ee.FeatureCollection(zoi_ndvi_joined.map(merge_zoi_ci_ndvi)) - - # ------------------------- - # STEP 3: Export or Push to GeoServer - # ------------------------- - layer_name = f"WaterRejapp_zoi_{proj_obj.name}_{proj_obj.id}" - sync_project_fc_to_geoserver(final_merged, proj_obj.name, layer_name, "waterrej") - - -def get_directory_size(path): - total_size = 0 - for dirpath, dirnames, filenames in os.walk(path): - for filename in filenames: - file_path = os.path.join(dirpath, filename) - if os.path.isfile(file_path): - total_size += os.path.getsize(file_path) - return total_size - - -def generate_geojson_with_ci_ndvi_ndmi( - zoi_asset, ci_asset, ndvi_asset, ndmi_asset, proj_id -): - # Load project - proj_obj = Project.objects.get(pk=proj_id) - - zoi = ee.FeatureCollection(zoi_asset) - print("Number of features zoi:", zoi.size().getInfo()) - - ci = ee.FeatureCollection(ci_asset) - print("Number of features zoi:", ci.size().getInfo()) - ndvi = ee.FeatureCollection(ndmi_asset) - print("Number of features zoi:", ndvi.size().getInfo()) - ndmi = ee.FeatureCollection(ndmi_asset) - print("Number of features zoi:", ndmi.size().getInfo()) - - # ------------------------- - # STEP 1: Join ZOI with CI - # ------------------------- - join = ee.Join.inner() - filter = ee.Filter.intersects(leftField=".geo", rightField=".geo") - zoi_ci_joined = join.apply(zoi, ci, filter) - - def merge_zoi_ci(pair): - zoi_feat = ee.Feature(pair.get("primary")) - ci_feat = ee.Feature(pair.get("secondary")) - merged_props = zoi_feat.toDictionary().combine(ci_feat.toDictionary(), True) - return ee.Feature(zoi_feat.geometry(), merged_props) # ✅ keep ZOI geom - - zoi_with_ci = ee.FeatureCollection(zoi_ci_joined.map(merge_zoi_ci)) - - # ------------------------- - # STEP 2: Join with NDVI - # ------------------------- - zoi_ndvi_joined = join.apply(zoi_with_ci, ndvi, filter) - - def merge_zoi_ci_ndvi(pair): - prev_feat = ee.Feature(pair.get("primary")) - ndvi_feat = ee.Feature(pair.get("secondary")) - merged_props = prev_feat.toDictionary().combine(ndvi_feat.toDictionary(), True) - return ee.Feature(prev_feat.geometry(), merged_props) # ✅ still ZOI geom - - zoi_ci_ndvi = ee.FeatureCollection(zoi_ndvi_joined.map(merge_zoi_ci_ndvi)) - - # ------------------------- - # STEP 3: Join with NDMI - # ------------------------- - zoi_ndmi_joined = join.apply(zoi_ci_ndvi, ndmi, filter) - - def merge_zoi_ci_ndvi_ndmi(pair): - prev_feat = ee.Feature(pair.get("primary")) - ndmi_feat = ee.Feature(pair.get("secondary")) - merged_props = prev_feat.toDictionary().combine(ndmi_feat.toDictionary(), True) - return ee.Feature(prev_feat.geometry(), merged_props) # ✅ keep ZOI geom - - final_merged = ee.FeatureCollection(zoi_ndmi_joined.map(merge_zoi_ci_ndvi_ndmi)) - - # ------------------------- - # STEP 4: Export or Push to GeoServer - # ------------------------- - layer_name = f"WaterRejapp_zoi_{proj_obj.name}_{proj_obj.id}" - print(layer_name) - sync_project_fc_to_geoserver(final_merged, proj_obj.name, layer_name, "waterrej") - - -def generate_geojson_with_ci_ndvi(zoi_asset, ci_asset, ndvi_asset, proj_id): - # Load project - proj_obj = Project.objects.get(pk=proj_id) - - # Initialize Earth Engine - ee_initialize(4) - - # Load FeatureCollections - zoi = ee.FeatureCollection(zoi_asset) - ci = ee.FeatureCollection(ci_asset) - ndvi = ee.FeatureCollection(ndvi_asset) - - print("ZOI:", zoi.size().getInfo()) - print("CI:", ci.size().getInfo()) - print("NDVI:", ndvi.size().getInfo()) - - # Common join logic on UID - join = ee.Join.inner() - uid_filter = ee.Filter.equals(leftField="UID", rightField="UID") - - # --- Join ZOI + CI --- - zoi_ci_joined = join.apply(zoi, ci, uid_filter) - - def merge_zoi_ci(pair): - zoi_feat = ee.Feature(pair.get("primary")) - ci_feat = ee.Feature(pair.get("secondary")) - merged_props = zoi_feat.toDictionary().combine(ci_feat.toDictionary(), True) - # Keep ZOI geometry only - return ee.Feature(zoi_feat.geometry(), merged_props) - - zoi_with_ci = ee.FeatureCollection(zoi_ci_joined.map(merge_zoi_ci)) - - # --- Join with NDVI --- - zoi_ndvi_joined = join.apply(zoi_with_ci, ndvi, uid_filter) - - def merge_with_ndvi(pair): - base_feat = ee.Feature(pair.get("primary")) - ndvi_feat = ee.Feature(pair.get("secondary")) - merged_props = base_feat.toDictionary().combine(ndvi_feat.toDictionary(), True) - # Always retain ZOI geometry - return ee.Feature(base_feat.geometry(), merged_props) - - merged_final = ee.FeatureCollection(zoi_ndvi_joined.map(merge_with_ndvi)) - - # --- Ensure ZOI geometry retained in all features --- - merged_final = merged_final.map( - lambda f: ee.Feature( - f.setGeometry( - ee.Feature( - zoi.filter(ee.Filter.eq("UID", f.get("UID"))).first() - ).geometry() - ) - ) - ) - - layer_name = f"WaterRejapp_zoi_{proj_obj.name}_{proj_obj.id}" - print(layer_name) - - sync_project_fc_to_geoserver(merged_final, proj_obj.name, layer_name, "waterrej") - - -def save_layer_info_to_db( - state, - district, - block, - layer_name, - asset_id, - dataset_name, - sync_to_geoserver=False, - layer_version="1.0", - algorithm=None, - algorithm_version="1.0", - misc=None, - is_override=False, - is_gee_asset=True, -): - print("inside the save_layer_info_to_db function") - - dataset = Dataset.objects.get(name=dataset_name) - - try: - state_obj = StateSOI.objects.get(state_name__iexact=state) - district_obj = DistrictSOI.objects.get( - district_name__iexact=district, state=state_obj - ) - block_obj = TehsilSOI.objects.get( - tehsil_name__iexact=block, district=district_obj - ) - except Exception as e: - print("Error fetching in state district block:", e) - return - - is_public = is_asset_public(asset_id) if is_gee_asset else False - - # Check if there’s an existing layer - existing_layer = ( - Layer.objects.filter( - dataset=dataset, - layer_name=layer_name, - state=state_obj, - district=district_obj, - block=block_obj, - ) - .order_by("-layer_version") - .first() - ) - - if existing_layer: - if existing_layer.algorithm_version != algorithm_version: - # Algorithm version changed --> create new record with incremented layer_version - new_layer_version = str(float(existing_layer.layer_version) + 1) - print( - f"Algorithm version changed. Creating new layer version: {new_layer_version}" - ) - layer_obj = Layer.objects.create( - dataset=dataset, - layer_name=layer_name, - state=state_obj, - district=district_obj, - block=block_obj, - layer_version=new_layer_version, - algorithm=algorithm, - algorithm_version=algorithm_version, - is_sync_to_geoserver=sync_to_geoserver, - is_public_gee_asset=is_public, - is_override=is_override, - misc=misc, - gee_asset_path=asset_id, - ) - else: - # Algorithm version is same --> update existing layer - print("Algorithm version same. Updating existing layer.") - for field, value in { - "algorithm": algorithm, - "algorithm_version": algorithm_version, - "is_sync_to_geoserver": sync_to_geoserver, - "is_public_gee_asset": is_public, - "is_override": is_override, - "misc": misc, - "gee_asset_path": asset_id, - }.items(): - setattr(existing_layer, field, value) - existing_layer.save() - layer_obj = existing_layer - else: - # No existing record --> create a new one - print("No existing layer found. Creating new one.") - layer_obj = Layer.objects.create( - dataset=dataset, - layer_name=layer_name, - state=state_obj, - district=district_obj, - block=block_obj, - layer_version=layer_version, - algorithm=algorithm, - algorithm_version=algorithm_version, - is_sync_to_geoserver=sync_to_geoserver, - is_public_gee_asset=is_public, - is_override=is_override, - misc=misc, - gee_asset_path=asset_id, - ) - - print(f"Saved layer info (id={layer_obj.id}, version={layer_obj.layer_version})") - return layer_obj.id - - -def get_existing_end_year(dataset_name, layer_name): - """fetch objects from db on the basis of dataset name and layer_name""" - dataset = Dataset.objects.get(name=dataset_name) - layer_obj = Layer.objects.get(dataset=dataset, layer_name=layer_name) - existing_end_date = layer_obj.misc["end_year"] - print("existing_end_date", existing_end_date) - return existing_end_date - - -def get_layer_object(state, district, block, layer_name, dataset_name): - state_obj = StateSOI.objects.get(state_name__iexact=state) - district_obj = DistrictSOI.objects.get( - district_name__iexact=district, state=state_obj - ) - block_obj = TehsilSOI.objects.get(tehsil_name__iexact=block, district=district_obj) - layer_obj = ( - Layer.objects.filter( - state=state_obj, - district=district_obj, - block=block_obj, - layer_name=layer_name, - dataset__name=dataset_name, - ) - .order_by("-layer_version") - .first() - ) - return layer_obj - - -def update_dashboard_geojson( - state=None, - district=None, - block=None, - layer_name=None, - workspace_name=None, - proj_id=None, -): - if state and block and block: - print(f"🔄 Updating GeoJSON for {state}, {district}, {block}") - - # Get related objects - state_obj = StateSOI.objects.get(state_name=state) - district_obj = DistrictSOI.objects.get(district_name=district) - tehsil_obj = TehsilSOI.objects.get(tehsil_name=block) # fixed typo - - # Get or create main record - obj, created = State_Disritct_Block_Properties.objects.get_or_create( - state=state_obj, district=district_obj, tehsil=tehsil_obj - ) - else: - obj = Project.objects.get(pk=proj_id) - - # Map suffix to json_key - suffix_to_key = { - "wb": "wb_geojson", - "zoi": "zoi_geojson", - "mws": "mws_geojson", - } - - # Detect which key this layer corresponds to - json_key = None - for suffix, key in suffix_to_key.items(): - if layer_name == f"{state}_{district}_{block}_{suffix}": - json_key = key - break - - if not json_key: - print(f"⚠️ Layer name {layer_name} did not match any known type.") - return - - # Construct GeoServer URL - waterrej_url = ( - f"https://geoserver.core-stack.org:8443/geoserver/waterrej/ows?" - f"service=WFS&version=1.0.0&request=GetFeature&typeName={workspace_name}:{layer_name}" - f"&outputFormat=application%2Fjson" - ) - - # Load existing dashboard_geojson or create new - if proj_id: - misc = obj.dashboard_geojson or {} - else: - misc = obj.geojson_path or {} - - # Ensure waterrej section exists - if "waterrej" not in misc: - misc["waterrej"] = {} - - # Update or add this specific json_key - misc["waterrej"][json_key] = waterrej_url - - # Save the updated JSON field - obj.dashboard_geojson = misc - obj.save() - - print(f"✅ Added/Updated {json_key} for {state}, {district}, {block}") - - -def clean_geometry(geom): - """ - Clean geometry: - - Dissolve multipolygon → single polygon - - Remove holes automatically - - Fix invalid topology - - Buffer tiny polygons - """ - - # 1. Dissolve multi-polygons and remove holes - geom = geom.dissolve(maxError=1) - - # 2. Fix invalid rings by simplifying slightly (NEVER buffer(0)) - geom = geom.simplify(1) - - # 3. Buffer polygons smaller than 1 pixel (< 900 m²) - area = geom.area() - geom = ee.Algorithms.If( - area.lt(900), - geom.buffer(15), - geom, # ensure raster pixel center is captured - ) - - return ee.Geometry(geom) - - -def safe_reduce_max(image, geom, scale=30): - geom = clean_geometry(geom) - - val = ( - image.unmask(0) - .reduceRegion( - reducer=ee.Reducer.max(), - geometry=geom, - scale=scale, - maxPixels=1e13, - tileScale=4, - bestEffort=True, - ) - .get("b1") - ) - - return ee.Number(ee.Algorithms.If(val, val, 0)) - - -# ------------------------------------------------------ -# SAFE REDUCE MAX FUNCTION -# ------------------------------------------------------ -def safe_reduce_max(image, geom, scale=30): - geom = clean_geometry(geom) - - result = ( - image.unmask(0) - .reduceRegion( - reducer=ee.Reducer.max(), - geometry=geom, - scale=scale, - maxPixels=1e13, - tileScale=4, - bestEffort=True, - ) - .get("b1") - ) - - # Convert null → 0 - return ee.Number(ee.Algorithms.If(result, result, 0)) - - -# ------------------------------------------------------ -# MAIN FUNCTION TO PROCESS SWB LAYER -# ------------------------------------------------------ -def generate_swb_layer_with_max_so_catchment( - roi=None, - app_type="MWS", - asset_suffix=None, - asset_folder=None, - gee_account_id=None, -): - ee_initialize(gee_account_id) - - # Build asset paths - base_path = get_gee_dir_path( - asset_folder, asset_path=GEE_PATHS[app_type]["GEE_ASSET_PATH"] - ) - - so_asset = f"{base_path}stream_order_{asset_suffix}_raster" - ca_asset = f"{base_path}catchment_area_{asset_suffix}_raster" - - # Load rasters - stream_order_band = ee.Image(so_asset).select("b1") - catchment_band = ee.Image(ca_asset).select("b1") - - # Processing per waterbody - def compute_for_feature(feature): - geom = feature.geometry() - - max_so = safe_reduce_max(stream_order_band, geom, scale=30) - max_ca = safe_reduce_max(catchment_band, geom, scale=30) - - return feature.set( - { - "max_stream_order": max_so, - "max_catchment_area": max_ca, - } - ) - - # Map over the feature collection - return roi.map(compute_for_feature) - - -def _get_prod_backend_url(): - return getattr(settings, "PROD_BACKEND_URL", "").rstrip("/") - - -def _get_prod_api_key(): - return getattr(settings, "PROD_BACKEND_API_KEY", "") - - -def _sync_layer_to_prod_db(payload: dict): - prod_url = _get_prod_backend_url() - if not prod_url: - return None - - endpoint = prod_url + "/api/v1/sync_layer_remote/" - try: - response = requests.post( - endpoint, - json=payload, - headers={"X-Api-Key": _get_prod_api_key()}, - timeout=30, - ) - if response.status_code not in (200, 201): - logger.warning( - "Prod DB sync returned %s for layer %s: %s", - response.status_code, - payload.get("layer_name"), - response.text, - ) - return None - layer_id = response.json().get("layer_id") - logger.info( - "Layer %s synced to prod DB (id=%s).", payload.get("layer_name"), layer_id - ) - return layer_id - except requests.RequestException as e: - logger.error( - "Failed to sync layer %s to prod DB: %s", payload.get("layer_name"), e - ) - return None - - -def _update_layer_sync_remote( - layer_id, sync_to_geoserver=None, is_stac_specs_generated=None -): - prod_url = _get_prod_backend_url() - if not prod_url or layer_id is None: - return - - endpoint = prod_url + "/api/v1/update_layer_sync_remote/" - payload = { - "layer_id": layer_id, - "sync_to_geoserver": sync_to_geoserver, - "is_stac_specs_generated": is_stac_specs_generated, - } - try: - response = requests.post( - endpoint, - json=payload, - headers={"X-Api-Key": _get_prod_api_key()}, - timeout=30, - ) - if response.status_code not in (200, 201): - logger.warning( - "Prod layer sync status update returned %s for layer %s: %s", - response.status_code, - layer_id, - response.text, - ) - else: - logger.info("Layer sync status updated on prod DB for id=%s.", layer_id) - except requests.RequestException as e: - logger.error( - "Failed to update layer sync status on prod DB for id=%s: %s", layer_id, e - ) - - -def update_layer_sync_status( - layer_id, sync_to_geoserver=None, is_stac_specs_generated=None -): - if _get_prod_backend_url(): - _update_layer_sync_remote( - layer_id, - sync_to_geoserver=sync_to_geoserver, - is_stac_specs_generated=is_stac_specs_generated, - ) - return layer_id - - try: - layer_obj = Layer.objects.filter(id=layer_id).first() - if layer_obj is None: - return None - - update_fields = [] - if sync_to_geoserver is not None: - layer_obj.is_sync_to_geoserver = sync_to_geoserver - update_fields.append("is_sync_to_geoserver") - if is_stac_specs_generated is not None: - layer_obj.is_stac_specs_generated = is_stac_specs_generated - update_fields.append("is_stac_specs_generated") - - # `save(update_fields=...)` fires the post_save signal so the STAC - # auto-trigger handler in `computing.signals` can pick up the flip. - if update_fields: - layer_obj.save(update_fields=update_fields) - print( - f"Updated {update_fields} for layer ID: {layer_id} " - f"(sync={sync_to_geoserver}, stac={is_stac_specs_generated})" - ) - return layer_id - - except Exception as e: - print(f"Error updating layer sync status: {e}") - - -def _is_cache_valid(cache: dict, workspace: str) -> bool: - if workspace not in cache: - return False - age = time.time() - cache[workspace]["cached_at"] - if age > 3600: - logger.info(f"Cache expired for {workspace} (age: {int(age)}s)") - return False - return True - - -def _set_cache(cache: dict, workspace: str, data: set): - cache[workspace] = { - "data": data, - "cached_at": time.time(), - } - - -def send_report_email( - result, - report_type: str = "missing_layers", - recipients: list = None, -) -> bool: - """ - Generic reusable function to email a JSON report. - report_type: "missing_layers" or "missing_excel_files" - """ - if recipients is None: - recipients = getattr(settings, "MISSING_LAYER_RECIPIENTS", []) - - if isinstance(recipients, str): - recipients = [recipients] - - if not recipients: - logger.error("No recipients configured for report email.") - return False - - if report_type == "missing_layers": - subject = "Missing Layers Report" - attachment_name = "missing_layers.json" - - strict_result = result.get("Mandatory", {}) - can_be_empty_result = result.get("can_be_empty", {}) - - # Filter out workspaces with zero missing layers - strict_filtered = { - ws: data for ws, data in strict_result.items() if data.get("missing_layers") - } - can_be_empty_filtered = { - ws: data - for ws, data in can_be_empty_result.items() - if data.get("missing_layers") - } - - strict_summary = [] - total_strict_missing = 0 - for layer, data in strict_filtered.items(): - count = len(data.get("missing_layers", [])) - total_strict_missing += count - strict_summary.append(f"{layer}: {count}") - - can_be_empty_summary = [] - total_can_be_empty_missing = 0 - for layer, data in can_be_empty_filtered.items(): - count = len(data.get("missing_layers", [])) - total_can_be_empty_missing += count - can_be_empty_summary.append(f"{layer}: {count}") - - total_missing = total_strict_missing + total_can_be_empty_missing - - body = ( - "Missing Layers Report\n\n" - f"Total Missing: {total_missing}\n" - f" - Mandatory (needs attention): {total_strict_missing}\n" - f" - Can-Be-Empty (may be legitimately absent): {total_can_be_empty_missing}\n\n" - "---- Mandatory Workspaces (data expected everywhere) ----\n" - + ( - "\n".join(strict_summary) - if strict_summary - else "None — nothing missing" - ) - + "\n\n" - "---- Can-Be-Empty Workspaces (some locations may legitimately have no data) ----\n" - + ( - "\n".join(can_be_empty_summary) - if can_be_empty_summary - else "None — nothing missing" - ) - + "\n\nDetailed report attached." - ) - result = { - "Mandatory": strict_filtered, - "can_be_empty": can_be_empty_filtered, - } - - elif report_type == "missing_excel_files": - subject = "Missing Stats Excel/JSON Files Report" - attachment_name = "missing_excel_files.json" - total_locations = len(result) - total_missing_files = sum(len(loc.get("missing_files", [])) for loc in result) - total_xlsx_issues = sum(1 for loc in result if loc.get("xlsx_issues")) - body = ( - "Missing Stats Excel/JSON Files Report\n\n" - f"Tehsils which files(json/excel) are missing: {total_locations}\n" - f"Total missing files: {total_missing_files}\n" - f"Tehsils with xlsx sheet missing: {total_xlsx_issues}\n\n" - "Detailed report attached." - ) - - else: - logger.error(f"Unknown report_type: {report_type}") - return False - - attachment_content = json.dumps(result, indent=4) - max_retries = 3 - - for attempt in range(max_retries): - connection = None - try: - connection = get_connection(timeout=120) - connection.open() - email = EmailMessage( - subject=subject, - body=body, - from_email=settings.EMAIL_HOST_USER, - to=recipients, - connection=connection, - ) - email.attach( - attachment_name, - attachment_content, - "application/json", - ) - email.send() - logger.info(f"{subject} sent to {recipients}") - logger.info( - f"Attachment size: " - f"{len(attachment_content.encode('utf-8')) / 1024:.2f} KB" - ) - return True - except Exception as e: - logger.exception(f"Attempt {attempt + 1}/{max_retries} failed: {e}") - if attempt < max_retries - 1: - wait_time = 5 * (attempt + 1) - logger.info(f"Retrying after {wait_time} seconds...") - time.sleep(wait_time) - else: - logger.error("All attempts to send email failed.") - return False - finally: - if connection: - try: - connection.close() - except Exception: - pass - return False +import copy +import json +import logging +import os +import shutil +import zipfile +from datetime import datetime, timedelta + +import ee +import fiona +import geopandas as gpd +import requests +from django.conf import settings +from shapely.geometry import shape +from shapely.validation import explain_validity + +from computing.models import Dataset, Layer +from geoadmin.models import ( + DistrictSOI, + State_Disritct_Block_Properties, + StateSOI, + TehsilSOI, +) +from projects.models import Project +from utilities.constants import ( + ADMIN_BOUNDARY_OUTPUT_DIR, + GEE_ASSET_PATH, + GEE_HELPER_PATH, + GEE_PATHS, + SHAPEFILE_DIR, +) +from utilities.gee_utils import ( + check_task_status, + ee_initialize, + get_gee_asset_path, + get_gee_dir_path, + get_geojson_from_gcs, + is_asset_public, + is_gee_asset_exists, + sync_vector_to_gcs, + valid_gee_text, +) +from utilities.geoserver_utils import Geoserver +from django.core.mail import EmailMessage, get_connection +import time + + +logger = logging.getLogger(__name__) + + +def generate_shape_files(path): + gdf = gpd.read_file(path + ".json") + if os.path.exists(path): + # Only replace the target shapefile directory. Removing the parent + # state/workspace directory here corrupts sibling outputs on reruns. + shutil.rmtree(path) + + os.makedirs(os.path.dirname(path), exist_ok=True) + gdf.to_file( + path, + driver="ESRI Shapefile", + ) + return path + + +def convert_to_zip(dir_name, file_type): + if file_type == "gpkg": + with zipfile.ZipFile(dir_name + ".zip", "w", zipfile.ZIP_DEFLATED) as zipf: + zipf.write(dir_name + ".gpkg", arcname=os.path.basename(dir_name + ".gpkg")) + return dir_name + ".zip" + else: + return shutil.make_archive(dir_name, "zip", dir_name + "/") + + +def push_shape_to_geoserver( + path, store_name=None, workspace=None, layer_name=None, file_type="shp" +): + geo = Geoserver() + + print(f"layer_name: {layer_name}") + if layer_name: + try: + print(f"Attempting to delete store: {layer_name}") + geo.delete_vector_store(workspace=workspace, store=layer_name) + print(f"Successfully deleted store: {layer_name}") + except Exception as e: + print(f"Store does not exist or error deleting: {str(e)}") + + zip_path = convert_to_zip(path, file_type) + print(f"Zip path: {zip_path}") + print(f"Store name: {store_name}") + print(f"Workspace: {workspace}") + + response = geo.create_shp_datastore( + path=zip_path, + store_name=store_name, + workspace=workspace, + file_extension=file_type, + ) + print(f"Response: {response}") + return response + + +def kml_to_geojson(state_name, district_name, block_name, kml_path): + fiona.drvsupport.supported_drivers["kml"] = ( + "rw" # enable KML support which is disabled by default + ) + fiona.drvsupport.supported_drivers["KML"] = ( + "rw" # enable KML support which is disabled by default + ) + gdf = gpd.read_file(kml_path) + geometry_types = gdf.geometry.geometry.type.unique() + state_dir = os.path.join(ADMIN_BOUNDARY_OUTPUT_DIR, state_name) + + for gtype in geometry_types: + df = gdf.loc[gdf.geometry.geometry.type == gtype] + path = os.path.join(state_dir, f"{district_name}_{block_name}_{gtype}") + df.to_file(path + ".json", driver="GeoJSON") + generate_shape_files(path) + push_shape_to_geoserver(path, workspace="test_workspace") + + +def convert_kml_to_shapefile(kml_path, output_dir, shapefile_name): + if not os.path.exists(output_dir + "/" + shapefile_name): + os.makedirs(output_dir + "/" + shapefile_name) + + shapefile_path = os.path.join( + output_dir + "/" + shapefile_name, shapefile_name + ".shp" + ) + print("path path", shapefile_path) + cmd = f"ogr2ogr -f 'ESRI Shapefile' {shapefile_path} {kml_path}" # output.shp input.kml + os.system(command=cmd) + + return output_dir + "/" + shapefile_name + + +def kml_to_shp(state_name, district_name, block_name, kml_path): + shapefile_name = f"{district_name}_{block_name}" + shapefile_layer_path = convert_kml_to_shapefile( + kml_path, SHAPEFILE_DIR, shapefile_name + ) + + push_shape_to_geoserver(shapefile_layer_path, workspace="customkml") + + # os.remove(kml_path) + # shutil.rmtree(shapefile_layer_path) + os.remove(shapefile_layer_path + ".zip") + + +def sync_layer_to_geoserver(state_name, fc, layer_name, workspace): + state_dir = os.path.join("data/fc_to_shape", state_name) + if not os.path.exists(state_dir): + os.mkdir(state_dir) + path = os.path.join(state_dir, f"{layer_name}") + # Write the feature collection into json file + with open(path + ".json", "w") as f: + try: + f.write(f"{json.dumps(fc)}") + except Exception as e: + print(e) + + path = generate_shape_files(path) + return push_shape_to_geoserver(path, workspace=workspace, layer_name=layer_name) + + +def sync_fc_to_geoserver(fc, shp_folder, layer_name, workspace, style_name=None): + try: + geojson_fc = fc.getInfo() + except Exception as e: + print("Exception in getInfo()", e) + task_id = sync_vector_to_gcs(fc, layer_name, "GeoJSON") + check_task_status([task_id]) + + geojson_fc = get_geojson_from_gcs(layer_name) + geo = Geoserver() + if len(geojson_fc["features"]) > 0: + state_dir = os.path.join("data/fc_to_shape", shp_folder) + if not os.path.exists(state_dir): + os.mkdir(state_dir) + path = os.path.join(state_dir, f"{layer_name}") + + # Convert to GeoDataFrame + gdf = gpd.GeoDataFrame.from_features(geojson_fc["features"]) + + # Set CRS (Earth Engine uses EPSG:4326 by default) + gdf.crs = "EPSG:4326" + + gdf = fix_invalid_geometry_in_gdf(gdf) + + # Save as GeoPackage + gdf.to_file(path + ".gpkg", driver="GPKG") + res = push_shape_to_geoserver(path, workspace=workspace, file_type="gpkg") + if style_name: + style_res = geo.publish_style( + layer_name=layer_name, style_name=style_name, workspace=workspace + ) + print("Style response:", style_res) + return res + else: + return "No features in FeatureCollection" + + +def sync_project_fc_to_geoserver(fc, project_name, layer_name, workspace): + print("inside") + print(layer_name) + try: + geojson_fc = fc.getInfo() + except Exception as e: + print("Exception in getInfo()", e) + task_id = sync_vector_to_gcs(fc, layer_name, "GeoJSON") + check_task_status([task_id]) + + geojson_fc = get_geojson_from_gcs(layer_name) + print(len(geojson_fc["features"])) + if len(geojson_fc["features"]) > 0: + state_dir = os.path.join("data/fc_to_shape", project_name) + if not os.path.exists(state_dir): + os.mkdir(state_dir) + path = os.path.join(state_dir, f"{layer_name}") + + # Convert to GeoDataFrame + gdf = gpd.GeoDataFrame.from_features(geojson_fc["features"]) + + # Set CRS (Earth Engine uses EPSG:4326 by default) + gdf.crs = "EPSG:4326" + + gdf = fix_invalid_geometry_in_gdf(gdf) + + # Save as GeoPackage + gdf.to_file(path + ".gpkg", driver="GPKG") + print("pushed to geoserver") + return push_shape_to_geoserver( + path, workspace=workspace, layer_name=layer_name, file_type="gpkg" + ) + else: + print("no features found") + return + + +def to_camelcase(text): + words = text.split() + camelcase = words[0].lower() + for word in words[1:]: + camelcase += word.capitalize() + return camelcase + + +def create_chunk(aoi, description, chunk_size): + size = aoi.size().getInfo() + parts = size // chunk_size + # task_ids = [] + rois = [] + descs = [] + for part in range(parts + 1): + start = part * chunk_size + end = start + chunk_size + block_name_for_parts = description + "_" + str(start) + "-" + str(end) + roi = ee.FeatureCollection(aoi.toList(aoi.size()).slice(start, end)) + if roi.size().getInfo() > 0: + descs.append(block_name_for_parts) + rois.append(roi) + + return rois, descs + + +def merge_chunks( + aoi, + folder_list, + description, + chunk_size, + chunk_asset_path=GEE_HELPER_PATH, + merge_asset_path=GEE_ASSET_PATH, + merge_asset_id=None, +): + print("Merge Chunk task initiated") + ee_initialize() + size = aoi.size().getInfo() + parts = size // chunk_size + assets = [] + for part in range(parts + 1): + start = part * chunk_size + end = start + chunk_size + block_name_for_parts = description + "_" + str(start) + "-" + str(end) + src_asset_id = ( + get_gee_dir_path(folder_list, chunk_asset_path) + block_name_for_parts + ) + if is_gee_asset_exists(src_asset_id): + assets.append(ee.FeatureCollection(src_asset_id)) + + asset = ee.FeatureCollection(assets).flatten() + + asset_id = merge_asset_id or ( + get_gee_dir_path(folder_list, merge_asset_path) + description + ) + try: + # Export an ee.FeatureCollection as an Earth Engine asset. + task = ee.batch.Export.table.toAsset( + **{ + "collection": asset, + "description": description, + "assetId": asset_id, + } + ) + + task.start() + print("Successfully started the merge chunk", task.status()) + return task.status()["id"] + except Exception as e: + print(f"Error occurred in running merge task: {e}") + return None + + +def fix_invalid_geometry_in_gdf(gdf): + invalid = gdf[~gdf.is_valid] + if not invalid.empty: + print("Invalid geometries found:") + for idx, geom in invalid.geometry.items(): + print(f"Index {idx}: {explain_validity(geom)}") + gdf.loc[idx, "geometry"] = gdf.loc[idx, "geometry"].buffer(0) + + return gdf + + +def get_season_key(date): + """Return season key like 'rabi_2017-2018' based on Indian cropping seasons.""" + month = date.month + year = date.year + next_year = year + 1 + + if month in [1, 2]: + return f"rabi_{year - 1}-{year}" # Jan–Feb → Rabi of previous year + elif month in [11, 12]: + return f"rabi_{year}-{next_year}" # Nov–Dec → Rabi starting this year + elif month in [3, 4, 5, 6]: + return f"zaid_{year}-{next_year}" + elif month in [7, 8, 9, 10]: + return f"kharif_{year}-{next_year}" + else: + return None + + +def get_agri_year_key(season_key): + """Convert a season key to agricultural year key (e.g., rabi_2017-2018 → 2017-2018).""" + season, years = season_key.split("_") + start_year, end_year = map(int, years.split("-")) + + if season in ["kharif", "rabi"]: + return f"{start_year}-{end_year}" + elif season == "zaid": + return f"{start_year - 1}-{start_year}" # Zaid 2018-2019 → Agri year 2017-2018 + else: + return None + + +def calculate_precipitation_season( + geojson_filepath, draught_asset_id, start_year=None, end_year=None +): + if start_year is None or end_year is None: + raise ValueError( + "start_year and end_year are required for calculate_precipitation_season." + ) + start_year = int(start_year) + end_year = int(end_year) + + # Load the GeoJSON file + with open(geojson_filepath, "r") as f: + feature_collection = json.load(f) + + features_ee = [] + + for feature in feature_collection["features"]: + original_props = feature["properties"] + new_props = {} + + # Copy UID + if "uid" in original_props: + new_props["uid"] = original_props["uid"] + + agri_year_totals = {} + + # Parse precipitation date keys + for key, val in original_props.items(): + try: + date = datetime.strptime(key, "%Y-%m-%d") + season_key = get_season_key(date) + if not season_key: + continue + + agri_key = get_agri_year_key(season_key) + if not agri_key: + continue + + agri_start = int(agri_key.split("-")[0]) + if not (start_year <= agri_start <= end_year): + continue + + season = season_key.split("_")[0] # kharif, rabi etc + full_key = f"{season}_{agri_key}" + + agri_year_totals[full_key] = agri_year_totals.get(full_key, 0) + float( + val + ) + + except Exception: + continue + + # Add all seasonal totals to new_props + for agri_key, total in agri_year_totals.items(): + new_props[f"precipitation_{agri_key}"] = total + + # Create EE Feature + geom_ee = ee.Geometry(feature["geometry"]) + feature_ee = ee.Feature(geom_ee, new_props) + features_ee.append(feature_ee) + + # Left side FC + mws_fc = ee.FeatureCollection(features_ee) + + return mws_fc + + +def generate_geojson_with_ci_and_ndvi(zoi_asset, ci_asset, ndvi_asset, proj_id): + # Load project + proj_obj = Project.objects.get(pk=proj_id) + + # Build CI and NDVI asset paths + asset_path_ci = ( + get_gee_dir_path( + [proj_obj.name], asset_path=GEE_PATHS["WATER_REJ"]["GEE_ASSET_PATH"] + ) + + ci_asset + ) + + asset_path_ndvi = ( + get_gee_dir_path( + [proj_obj.name], asset_path=GEE_PATHS["WATER_REJ"]["GEE_ASSET_PATH"] + ) + + ndvi_asset + ) + + # Load FeatureCollections + zoi = ee.FeatureCollection(zoi_asset) + ci = ee.FeatureCollection(asset_path_ci) + ndvi = ee.FeatureCollection(asset_path_ndvi) + + # ------------------------- + # STEP 1: Join ZOI with Cropping Intensity + # ------------------------- + join = ee.Join.inner() + filter = ee.Filter.intersects(leftField=".geo", rightField=".geo") + zoi_ci_joined = join.apply(zoi, ci, filter) + + def merge_zoi_ci(pair): + zoi_feat = ee.Feature(pair.get("primary")) + ci_feat = ee.Feature(pair.get("secondary")) + merged_props = zoi_feat.toDictionary().combine(ci_feat.toDictionary(), True) + return ee.Feature(zoi_feat.geometry(), merged_props) + + zoi_with_ci = ee.FeatureCollection(zoi_ci_joined.map(merge_zoi_ci)) + + # ------------------------- + # STEP 2: Join ZOI+CI with NDVI + # ------------------------- + zoi_ndvi_joined = join.apply(zoi_with_ci, ndvi, filter) + + def merge_zoi_ci_ndvi(pair): + ci_feat = ee.Feature(pair.get("primary")) + ndvi_feat = ee.Feature(pair.get("secondary")) + merged_props = ci_feat.toDictionary().combine(ndvi_feat.toDictionary(), True) + return ee.Feature(ci_feat.geometry(), merged_props) + + final_merged = ee.FeatureCollection(zoi_ndvi_joined.map(merge_zoi_ci_ndvi)) + + # ------------------------- + # STEP 3: Export or Push to GeoServer + # ------------------------- + layer_name = f"WaterRejapp_zoi_{proj_obj.name}_{proj_obj.id}" + sync_project_fc_to_geoserver(final_merged, proj_obj.name, layer_name, "waterrej") + + +def get_directory_size(path): + total_size = 0 + for dirpath, dirnames, filenames in os.walk(path): + for filename in filenames: + file_path = os.path.join(dirpath, filename) + if os.path.isfile(file_path): + total_size += os.path.getsize(file_path) + return total_size + + +def generate_geojson_with_ci_ndvi_ndmi( + zoi_asset, ci_asset, ndvi_asset, ndmi_asset, proj_id +): + # Load project + proj_obj = Project.objects.get(pk=proj_id) + + zoi = ee.FeatureCollection(zoi_asset) + print("Number of features zoi:", zoi.size().getInfo()) + + ci = ee.FeatureCollection(ci_asset) + print("Number of features zoi:", ci.size().getInfo()) + ndvi = ee.FeatureCollection(ndmi_asset) + print("Number of features zoi:", ndvi.size().getInfo()) + ndmi = ee.FeatureCollection(ndmi_asset) + print("Number of features zoi:", ndmi.size().getInfo()) + + # ------------------------- + # STEP 1: Join ZOI with CI + # ------------------------- + join = ee.Join.inner() + filter = ee.Filter.intersects(leftField=".geo", rightField=".geo") + zoi_ci_joined = join.apply(zoi, ci, filter) + + def merge_zoi_ci(pair): + zoi_feat = ee.Feature(pair.get("primary")) + ci_feat = ee.Feature(pair.get("secondary")) + merged_props = zoi_feat.toDictionary().combine(ci_feat.toDictionary(), True) + return ee.Feature(zoi_feat.geometry(), merged_props) # ✅ keep ZOI geom + + zoi_with_ci = ee.FeatureCollection(zoi_ci_joined.map(merge_zoi_ci)) + + # ------------------------- + # STEP 2: Join with NDVI + # ------------------------- + zoi_ndvi_joined = join.apply(zoi_with_ci, ndvi, filter) + + def merge_zoi_ci_ndvi(pair): + prev_feat = ee.Feature(pair.get("primary")) + ndvi_feat = ee.Feature(pair.get("secondary")) + merged_props = prev_feat.toDictionary().combine(ndvi_feat.toDictionary(), True) + return ee.Feature(prev_feat.geometry(), merged_props) # ✅ still ZOI geom + + zoi_ci_ndvi = ee.FeatureCollection(zoi_ndvi_joined.map(merge_zoi_ci_ndvi)) + + # ------------------------- + # STEP 3: Join with NDMI + # ------------------------- + zoi_ndmi_joined = join.apply(zoi_ci_ndvi, ndmi, filter) + + def merge_zoi_ci_ndvi_ndmi(pair): + prev_feat = ee.Feature(pair.get("primary")) + ndmi_feat = ee.Feature(pair.get("secondary")) + merged_props = prev_feat.toDictionary().combine(ndmi_feat.toDictionary(), True) + return ee.Feature(prev_feat.geometry(), merged_props) # ✅ keep ZOI geom + + final_merged = ee.FeatureCollection(zoi_ndmi_joined.map(merge_zoi_ci_ndvi_ndmi)) + + # ------------------------- + # STEP 4: Export or Push to GeoServer + # ------------------------- + layer_name = f"WaterRejapp_zoi_{proj_obj.name}_{proj_obj.id}" + print(layer_name) + sync_project_fc_to_geoserver(final_merged, proj_obj.name, layer_name, "waterrej") + + +def generate_geojson_with_ci_ndvi(zoi_asset, ci_asset, ndvi_asset, proj_id): + # Load project + proj_obj = Project.objects.get(pk=proj_id) + + # Initialize Earth Engine + ee_initialize(4) + + # Load FeatureCollections + zoi = ee.FeatureCollection(zoi_asset) + ci = ee.FeatureCollection(ci_asset) + ndvi = ee.FeatureCollection(ndvi_asset) + + print("ZOI:", zoi.size().getInfo()) + print("CI:", ci.size().getInfo()) + print("NDVI:", ndvi.size().getInfo()) + + # Common join logic on UID + join = ee.Join.inner() + uid_filter = ee.Filter.equals(leftField="UID", rightField="UID") + + # --- Join ZOI + CI --- + zoi_ci_joined = join.apply(zoi, ci, uid_filter) + + def merge_zoi_ci(pair): + zoi_feat = ee.Feature(pair.get("primary")) + ci_feat = ee.Feature(pair.get("secondary")) + merged_props = zoi_feat.toDictionary().combine(ci_feat.toDictionary(), True) + # Keep ZOI geometry only + return ee.Feature(zoi_feat.geometry(), merged_props) + + zoi_with_ci = ee.FeatureCollection(zoi_ci_joined.map(merge_zoi_ci)) + + # --- Join with NDVI --- + zoi_ndvi_joined = join.apply(zoi_with_ci, ndvi, uid_filter) + + def merge_with_ndvi(pair): + base_feat = ee.Feature(pair.get("primary")) + ndvi_feat = ee.Feature(pair.get("secondary")) + merged_props = base_feat.toDictionary().combine(ndvi_feat.toDictionary(), True) + # Always retain ZOI geometry + return ee.Feature(base_feat.geometry(), merged_props) + + merged_final = ee.FeatureCollection(zoi_ndvi_joined.map(merge_with_ndvi)) + + # --- Ensure ZOI geometry retained in all features --- + merged_final = merged_final.map( + lambda f: ee.Feature( + f.setGeometry( + ee.Feature( + zoi.filter(ee.Filter.eq("UID", f.get("UID"))).first() + ).geometry() + ) + ) + ) + + layer_name = f"WaterRejapp_zoi_{proj_obj.name}_{proj_obj.id}" + print(layer_name) + + sync_project_fc_to_geoserver(merged_final, proj_obj.name, layer_name, "waterrej") + + +def save_layer_info_to_db( + state, + district, + block, + layer_name, + asset_id, + dataset_name, + sync_to_geoserver=False, + layer_version="1.0", + algorithm=None, + algorithm_version="1.0", + misc=None, + is_override=False, + is_gee_asset=True, +): + print("inside the save_layer_info_to_db function") + + dataset = Dataset.objects.get(name=dataset_name) + + try: + state_obj = StateSOI.objects.get(state_name__iexact=state) + district_obj = DistrictSOI.objects.get( + district_name__iexact=district, state=state_obj + ) + block_obj = TehsilSOI.objects.get( + tehsil_name__iexact=block, district=district_obj + ) + except Exception as e: + print("Error fetching in state district block:", e) + return + + is_public = is_asset_public(asset_id) if is_gee_asset else False + + # Check if there’s an existing layer + existing_layer = ( + Layer.objects.filter( + dataset=dataset, + layer_name=layer_name, + state=state_obj, + district=district_obj, + block=block_obj, + ) + .order_by("-layer_version") + .first() + ) + + if existing_layer: + if existing_layer.algorithm_version != algorithm_version: + # Algorithm version changed --> create new record with incremented layer_version + new_layer_version = str(float(existing_layer.layer_version) + 1) + print( + f"Algorithm version changed. Creating new layer version: {new_layer_version}" + ) + layer_obj = Layer.objects.create( + dataset=dataset, + layer_name=layer_name, + state=state_obj, + district=district_obj, + block=block_obj, + layer_version=new_layer_version, + algorithm=algorithm, + algorithm_version=algorithm_version, + is_sync_to_geoserver=sync_to_geoserver, + is_public_gee_asset=is_public, + is_override=is_override, + misc=misc, + gee_asset_path=asset_id, + ) + else: + # Algorithm version is same --> update existing layer + print("Algorithm version same. Updating existing layer.") + for field, value in { + "algorithm": algorithm, + "algorithm_version": algorithm_version, + "is_sync_to_geoserver": sync_to_geoserver, + "is_public_gee_asset": is_public, + "is_override": is_override, + "misc": misc, + "gee_asset_path": asset_id, + }.items(): + setattr(existing_layer, field, value) + existing_layer.save() + layer_obj = existing_layer + else: + # No existing record --> create a new one + print("No existing layer found. Creating new one.") + layer_obj = Layer.objects.create( + dataset=dataset, + layer_name=layer_name, + state=state_obj, + district=district_obj, + block=block_obj, + layer_version=layer_version, + algorithm=algorithm, + algorithm_version=algorithm_version, + is_sync_to_geoserver=sync_to_geoserver, + is_public_gee_asset=is_public, + is_override=is_override, + misc=misc, + gee_asset_path=asset_id, + ) + + print(f"Saved layer info (id={layer_obj.id}, version={layer_obj.layer_version})") + return layer_obj.id + + +def get_existing_end_year(dataset_name, layer_name): + """fetch objects from db on the basis of dataset name and layer_name""" + dataset = Dataset.objects.get(name=dataset_name) + layer_obj = Layer.objects.get(dataset=dataset, layer_name=layer_name) + existing_end_date = layer_obj.misc["end_year"] + print("existing_end_date", existing_end_date) + return existing_end_date + + +def get_layer_object(state, district, block, layer_name, dataset_name): + state_obj = StateSOI.objects.get(state_name__iexact=state) + district_obj = DistrictSOI.objects.get( + district_name__iexact=district, state=state_obj + ) + block_obj = TehsilSOI.objects.get(tehsil_name__iexact=block, district=district_obj) + layer_obj = ( + Layer.objects.filter( + state=state_obj, + district=district_obj, + block=block_obj, + layer_name=layer_name, + dataset__name=dataset_name, + ) + .order_by("-layer_version") + .first() + ) + return layer_obj + + +def update_dashboard_geojson( + state=None, + district=None, + block=None, + layer_name=None, + workspace_name=None, + proj_id=None, +): + if state and block and block: + print(f"🔄 Updating GeoJSON for {state}, {district}, {block}") + + # Get related objects + state_obj = StateSOI.objects.get(state_name=state) + district_obj = DistrictSOI.objects.get(district_name=district) + tehsil_obj = TehsilSOI.objects.get(tehsil_name=block) # fixed typo + + # Get or create main record + obj, created = State_Disritct_Block_Properties.objects.get_or_create( + state=state_obj, district=district_obj, tehsil=tehsil_obj + ) + else: + obj = Project.objects.get(pk=proj_id) + + # Map suffix to json_key + suffix_to_key = { + "wb": "wb_geojson", + "zoi": "zoi_geojson", + "mws": "mws_geojson", + } + + # Detect which key this layer corresponds to + json_key = None + for suffix, key in suffix_to_key.items(): + if layer_name == f"{state}_{district}_{block}_{suffix}": + json_key = key + break + + if not json_key: + print(f"⚠️ Layer name {layer_name} did not match any known type.") + return + + # Construct GeoServer URL + waterrej_url = ( + f"https://geoserver.core-stack.org:8443/geoserver/waterrej/ows?" + f"service=WFS&version=1.0.0&request=GetFeature&typeName={workspace_name}:{layer_name}" + f"&outputFormat=application%2Fjson" + ) + + # Load existing dashboard_geojson or create new + if proj_id: + misc = obj.dashboard_geojson or {} + else: + misc = obj.geojson_path or {} + + # Ensure waterrej section exists + if "waterrej" not in misc: + misc["waterrej"] = {} + + # Update or add this specific json_key + misc["waterrej"][json_key] = waterrej_url + + # Save the updated JSON field + obj.dashboard_geojson = misc + obj.save() + + print(f"✅ Added/Updated {json_key} for {state}, {district}, {block}") + + +def clean_geometry(geom): + """ + Clean geometry: + - Dissolve multipolygon → single polygon + - Remove holes automatically + - Fix invalid topology + - Buffer tiny polygons + """ + + # 1. Dissolve multi-polygons and remove holes + geom = geom.dissolve(maxError=1) + + # 2. Fix invalid rings by simplifying slightly (NEVER buffer(0)) + geom = geom.simplify(1) + + # 3. Buffer polygons smaller than 1 pixel (< 900 m²) + area = geom.area() + geom = ee.Algorithms.If( + area.lt(900), + geom.buffer(15), + geom, # ensure raster pixel center is captured + ) + + return ee.Geometry(geom) + + +def safe_reduce_max(image, geom, scale=30): + geom = clean_geometry(geom) + + val = ( + image.unmask(0) + .reduceRegion( + reducer=ee.Reducer.max(), + geometry=geom, + scale=scale, + maxPixels=1e13, + tileScale=4, + bestEffort=True, + ) + .get("b1") + ) + + return ee.Number(ee.Algorithms.If(val, val, 0)) + + +# ------------------------------------------------------ +# SAFE REDUCE MAX FUNCTION +# ------------------------------------------------------ +def safe_reduce_max(image, geom, scale=30): + geom = clean_geometry(geom) + + result = ( + image.unmask(0) + .reduceRegion( + reducer=ee.Reducer.max(), + geometry=geom, + scale=scale, + maxPixels=1e13, + tileScale=4, + bestEffort=True, + ) + .get("b1") + ) + + # Convert null → 0 + return ee.Number(ee.Algorithms.If(result, result, 0)) + + +# ------------------------------------------------------ +# MAIN FUNCTION TO PROCESS SWB LAYER +# ------------------------------------------------------ +def generate_swb_layer_with_max_so_catchment( + roi=None, + app_type="MWS", + asset_suffix=None, + asset_folder=None, + gee_account_id=None, +): + ee_initialize(gee_account_id) + + # Build asset paths + base_path = get_gee_dir_path( + asset_folder, asset_path=GEE_PATHS[app_type]["GEE_ASSET_PATH"] + ) + + so_asset = f"{base_path}stream_order_{asset_suffix}_raster" + ca_asset = f"{base_path}catchment_area_{asset_suffix}_raster" + + # Load rasters + stream_order_band = ee.Image(so_asset).select("b1") + catchment_band = ee.Image(ca_asset).select("b1") + + # Processing per waterbody + def compute_for_feature(feature): + geom = feature.geometry() + + max_so = safe_reduce_max(stream_order_band, geom, scale=30) + max_ca = safe_reduce_max(catchment_band, geom, scale=30) + + return feature.set( + { + "max_stream_order": max_so, + "max_catchment_area": max_ca, + } + ) + + # Map over the feature collection + return roi.map(compute_for_feature) + + +def _get_prod_backend_url(): + return getattr(settings, "PROD_BACKEND_URL", "").rstrip("/") + + +def _get_prod_api_key(): + return getattr(settings, "PROD_BACKEND_API_KEY", "") + + +def _sync_layer_to_prod_db(payload: dict): + prod_url = _get_prod_backend_url() + if not prod_url: + return None + + endpoint = prod_url + "/api/v1/sync_layer_remote/" + try: + response = requests.post( + endpoint, + json=payload, + headers={"X-Api-Key": _get_prod_api_key()}, + timeout=30, + ) + if response.status_code not in (200, 201): + logger.warning( + "Prod DB sync returned %s for layer %s: %s", + response.status_code, + payload.get("layer_name"), + response.text, + ) + return None + layer_id = response.json().get("layer_id") + logger.info( + "Layer %s synced to prod DB (id=%s).", payload.get("layer_name"), layer_id + ) + return layer_id + except requests.RequestException as e: + logger.error( + "Failed to sync layer %s to prod DB: %s", payload.get("layer_name"), e + ) + return None + + +def _update_layer_sync_remote( + layer_id, sync_to_geoserver=None, is_stac_specs_generated=None +): + prod_url = _get_prod_backend_url() + if not prod_url or layer_id is None: + return + + endpoint = prod_url + "/api/v1/update_layer_sync_remote/" + payload = { + "layer_id": layer_id, + "sync_to_geoserver": sync_to_geoserver, + "is_stac_specs_generated": is_stac_specs_generated, + } + try: + response = requests.post( + endpoint, + json=payload, + headers={"X-Api-Key": _get_prod_api_key()}, + timeout=30, + ) + if response.status_code not in (200, 201): + logger.warning( + "Prod layer sync status update returned %s for layer %s: %s", + response.status_code, + layer_id, + response.text, + ) + else: + logger.info("Layer sync status updated on prod DB for id=%s.", layer_id) + except requests.RequestException as e: + logger.error( + "Failed to update layer sync status on prod DB for id=%s: %s", layer_id, e + ) + + +def update_layer_sync_status( + layer_id, sync_to_geoserver=None, is_stac_specs_generated=None +): + if _get_prod_backend_url(): + _update_layer_sync_remote( + layer_id, + sync_to_geoserver=sync_to_geoserver, + is_stac_specs_generated=is_stac_specs_generated, + ) + return layer_id + + try: + layer_obj = Layer.objects.filter(id=layer_id).first() + if layer_obj is None: + return None + + update_fields = [] + if sync_to_geoserver is not None: + layer_obj.is_sync_to_geoserver = sync_to_geoserver + update_fields.append("is_sync_to_geoserver") + if is_stac_specs_generated is not None: + layer_obj.is_stac_specs_generated = is_stac_specs_generated + update_fields.append("is_stac_specs_generated") + + # `save(update_fields=...)` fires the post_save signal so the STAC + # auto-trigger handler in `computing.signals` can pick up the flip. + if update_fields: + layer_obj.save(update_fields=update_fields) + print( + f"Updated {update_fields} for layer ID: {layer_id} " + f"(sync={sync_to_geoserver}, stac={is_stac_specs_generated})" + ) + return layer_id + + except Exception as e: + print(f"Error updating layer sync status: {e}") + + +def _is_cache_valid(cache: dict, workspace: str) -> bool: + if workspace not in cache: + return False + age = time.time() - cache[workspace]["cached_at"] + if age > 3600: + logger.info(f"Cache expired for {workspace} (age: {int(age)}s)") + return False + return True + + +def _set_cache(cache: dict, workspace: str, data: set): + cache[workspace] = { + "data": data, + "cached_at": time.time(), + } + + +def send_report_email( + result, + report_type: str = "missing_layers", + recipients: list = None, +) -> bool: + """ + Generic reusable function to email a JSON report. + report_type: "missing_layers" or "missing_excel_files" + """ + if recipients is None: + recipients = getattr(settings, "MISSING_LAYER_RECIPIENTS", []) + + if isinstance(recipients, str): + recipients = [recipients] + + if not recipients: + logger.error("No recipients configured for report email.") + return False + + if report_type == "missing_layers": + subject = "Missing Layers Report" + attachment_name = "missing_layers.json" + + strict_result = result.get("Mandatory", {}) + can_be_empty_result = result.get("can_be_empty", {}) + + # Filter out workspaces with zero missing layers + strict_filtered = { + ws: data for ws, data in strict_result.items() if data.get("missing_layers") + } + can_be_empty_filtered = { + ws: data + for ws, data in can_be_empty_result.items() + if data.get("missing_layers") + } + + strict_summary = [] + total_strict_missing = 0 + for layer, data in strict_filtered.items(): + count = len(data.get("missing_layers", [])) + total_strict_missing += count + strict_summary.append(f"{layer}: {count}") + + can_be_empty_summary = [] + total_can_be_empty_missing = 0 + for layer, data in can_be_empty_filtered.items(): + count = len(data.get("missing_layers", [])) + total_can_be_empty_missing += count + can_be_empty_summary.append(f"{layer}: {count}") + + total_missing = total_strict_missing + total_can_be_empty_missing + + body = ( + "Missing Layers Report\n\n" + f"Total Missing: {total_missing}\n" + f" - Mandatory (needs attention): {total_strict_missing}\n" + f" - Can-Be-Empty (may be legitimately absent): {total_can_be_empty_missing}\n\n" + "---- Mandatory Workspaces (data expected everywhere) ----\n" + + ( + "\n".join(strict_summary) + if strict_summary + else "None — nothing missing" + ) + + "\n\n" + "---- Can-Be-Empty Workspaces (some locations may legitimately have no data) ----\n" + + ( + "\n".join(can_be_empty_summary) + if can_be_empty_summary + else "None — nothing missing" + ) + + "\n\nDetailed report attached." + ) + result = { + "Mandatory": strict_filtered, + "can_be_empty": can_be_empty_filtered, + } + + elif report_type == "missing_excel_files": + subject = "Missing Stats Excel/JSON Files Report" + attachment_name = "missing_excel_files.json" + total_locations = len(result) + total_missing_files = sum(len(loc.get("missing_files", [])) for loc in result) + total_xlsx_issues = sum(1 for loc in result if loc.get("xlsx_issues")) + body = ( + "Missing Stats Excel/JSON Files Report\n\n" + f"Tehsils which files(json/excel) are missing: {total_locations}\n" + f"Total missing files: {total_missing_files}\n" + f"Tehsils with xlsx sheet missing: {total_xlsx_issues}\n\n" + "Detailed report attached." + ) + + else: + logger.error(f"Unknown report_type: {report_type}") + return False + + attachment_content = json.dumps(result, indent=4) + max_retries = 3 + + for attempt in range(max_retries): + connection = None + try: + connection = get_connection(timeout=120) + connection.open() + email = EmailMessage( + subject=subject, + body=body, + from_email=settings.EMAIL_HOST_USER, + to=recipients, + connection=connection, + ) + email.attach( + attachment_name, + attachment_content, + "application/json", + ) + email.send() + logger.info(f"{subject} sent to {recipients}") + logger.info( + f"Attachment size: " + f"{len(attachment_content.encode('utf-8')) / 1024:.2f} KB" + ) + return True + except Exception as e: + logger.exception(f"Attempt {attempt + 1}/{max_retries} failed: {e}") + if attempt < max_retries - 1: + wait_time = 5 * (attempt + 1) + logger.info(f"Retrying after {wait_time} seconds...") + time.sleep(wait_time) + else: + logger.error("All attempts to send email failed.") + return False + finally: + if connection: + try: + connection.close() + except Exception: + pass + return False diff --git a/computing/zoi_layers/zoi.py b/computing/zoi_layers/zoi.py index b8be98b2..5a1fcc17 100644 --- a/computing/zoi_layers/zoi.py +++ b/computing/zoi_layers/zoi.py @@ -1,10 +1,47 @@ +from datetime import datetime + from computing.zoi_layers.zoi1 import generate_zoi1 from computing.zoi_layers.zoi2 import generate_zoi_ci from computing.zoi_layers.zoi3 import get_ndvi_for_zoi -from projects.models import Project -from utilities.gee_utils import ee_initialize, valid_gee_text, check_task_status -from waterrejuvenation.utils import wait_for_task_completion, delete_asset_on_GEE from nrm_app.celery import app +from projects.models import Project +from utilities.gee_utils import ee_initialize, valid_gee_text + + +def _resolve_zoi_time_window(start_date=None, end_date=None): + """ + Validate and normalize ZOI date window. + + Requires explicit start_date and end_date (YYYY-MM-DD). No year defaults — + callers must provide the analysis window. + """ + if not start_date or not end_date: + raise ValueError( + "start_date and end_date are required (YYYY-MM-DD). " + "Pass both parameters explicitly; no default date window is applied." + ) + + start_date = str(start_date).strip() + end_date = str(end_date).strip() + + try: + start_dt = datetime.strptime(start_date, "%Y-%m-%d") + end_dt = datetime.strptime(end_date, "%Y-%m-%d") + except ValueError as exc: + raise ValueError("start_date and end_date must be in YYYY-MM-DD format.") from exc + + if start_dt > end_dt: + raise ValueError("start_date must be less than or equal to end_date.") + + # ZOI CI/NDVI use hydrological years (July -> June), represented by start-year. + start_year = start_dt.year if start_dt.month >= 7 else start_dt.year - 1 + end_year = end_dt.year if end_dt.month >= 7 else end_dt.year - 1 + if start_year > end_year: + raise ValueError( + "Provided date window does not contain a valid hydrological year." + ) + + return start_date, end_date, start_year, end_year @app.task() @@ -18,6 +55,8 @@ def generate_zoi( app_type="MWS", gee_account_id=None, proj_id=None, + start_date=None, + end_date=None, ): print(f"gee account id {gee_account_id}") ee_initialize(gee_account_id) @@ -31,6 +70,10 @@ def generate_zoi( asset_folder_list = [proj_obj.name.lower()] asset_suffix = f"{proj_obj.name}_{proj_obj.id}".lower() + start_date, end_date, start_year, end_year = _resolve_zoi_time_window( + start_date, end_date + ) + generate_zoi1( state, district, @@ -41,6 +84,8 @@ def generate_zoi( app_type, gee_account_id, proj_id, + start_date=start_date, + end_date=end_date, ) generate_zoi_ci( @@ -52,10 +97,13 @@ def generate_zoi( app_type, gee_account_id, proj_id, + start_date=start_date, + end_date=end_date, + start_year=start_year, + end_year=end_year, ) if proj_id: - get_ndvi_for_zoi( state=state, district=district, @@ -65,4 +113,8 @@ def generate_zoi( app_type=app_type, gee_account_id=gee_account_id, proj_id=proj_id, + start_date=start_date, + end_date=end_date, + start_year=start_year, + end_year=end_year, ) diff --git a/computing/zoi_layers/zoi1.py b/computing/zoi_layers/zoi1.py index 80670092..3f89d367 100644 --- a/computing/zoi_layers/zoi1.py +++ b/computing/zoi_layers/zoi1.py @@ -25,7 +25,6 @@ ee_initialize, valid_gee_text, get_gee_dir_path, - is_gee_asset_exists, export_vector_asset_to_gee, make_asset_public, check_task_status, @@ -36,6 +35,7 @@ calculate_zoi_area, wait_for_task_completion, delete_asset_on_GEE, + _waterbody_area_ha, ) from computing.surface_water_bodies.swb import sync_asset_to_db_and_geoserver @@ -50,8 +50,14 @@ def generate_zoi1( app_type="MWS", gee_account_id=None, proj_id=None, + start_date=None, + end_date=None, ): print("insdie zoi") + if not start_date or not end_date: + raise ValueError( + "start_date and end_date are required for ZOI generation (YYYY-MM-DD)." + ) ee_initialize(gee_account_id) description = "swb3_" + asset_suffix asset_id = ( @@ -74,15 +80,12 @@ def generate_zoi1( + description_zoi ) delete_asset_on_GEE(asset_id_zoi) - start_date = "2017-07-01" - end_date = "2025-06-30" zoi_fc = roi.map(compute_zoi) zoi_fc = ee.FeatureCollection(zoi_fc) zoi_rings = zoi_fc.filter(ee.Filter.gt("zoi_wb", 0)).map(create_ring) - if not is_gee_asset_exists(asset_id_zoi): - zoi_task = export_vector_asset_to_gee(zoi_rings, description_zoi, asset_id_zoi) - check_task_status([zoi_task]) - make_asset_public(asset_id_zoi) + zoi_task = export_vector_asset_to_gee(zoi_rings, description_zoi, asset_id_zoi) + check_task_status([zoi_task]) + make_asset_public(asset_id_zoi) if state and district and block: layer_name = f"waterbodies_zoi_{asset_suffix}" print(layer_name) @@ -103,13 +106,6 @@ def generate_zoi1( sync_project_fc_to_geoserver(zoi_rings, proj_obj.name, layer_name, "zoi_layers") -def _waterbody_area_ha(feature): - """Area in hectares; use geometry when area_ored is missing (e.g. water rej layers).""" - geom_area_ha = ee.Number(feature.geometry().area(maxError=1)).divide(10000) - stored_area = feature.get("area_ored") - return ee.Number(ee.Algorithms.If(stored_area, stored_area, geom_area_ha)) - - def compute_zoi(feature): area_of_wb = _waterbody_area_ha(feature) @@ -136,13 +132,32 @@ def y_large_bodies(area): .add(s.multiply(y_large_bodies(area_of_wb)).round()) ) - return feature.set("zoi_wb", zoi) + return feature.set("zoi_wb", zoi).set( + "UID", + ee.Algorithms.If( + feature.get("UID"), + feature.get("UID"), + ee.Algorithms.If( + feature.get("uid"), + feature.get("uid"), + feature.get("MWS_UID"), + ), + ), + ) def create_ring(feature): geom = feature.geometry() # can be point or polygon zoi = ee.Number(feature.get("zoi_wb")) - uid = feature.get("UID") + uid = ee.Algorithms.If( + feature.get("UID"), + feature.get("UID"), + ee.Algorithms.If( + feature.get("uid"), + feature.get("uid"), + feature.get("MWS_UID"), + ), + ) # Make circle buffer from centroid centroid = geom.centroid() diff --git a/computing/zoi_layers/zoi2.py b/computing/zoi_layers/zoi2.py index 03d44196..c7e44946 100644 --- a/computing/zoi_layers/zoi2.py +++ b/computing/zoi_layers/zoi2.py @@ -19,11 +19,20 @@ def generate_zoi_ci( gee_account_id=None, proj_id=None, roi=None, + start_date=None, + end_date=None, + start_year=None, + end_year=None, ): from computing.cropping_intensity.cropping_intensity import ( generate_cropping_intensity, ) + if not start_date or not end_date or start_year is None or end_year is None: + raise ValueError( + "start_date, end_date, start_year, and end_year are required for ZOI CI." + ) + if state and district and block: asset_suffix = ( valid_gee_text(district.lower()) + "_" + valid_gee_text(block.lower()) @@ -46,6 +55,14 @@ def generate_zoi_ci( + description_ci ) delete_asset_on_GEE(asset_id_ci) + description_zoi_ci = f"cropping_intensity_zoi_{asset_suffix}" + asset_id_zoi_ci = ( + get_gee_dir_path( + asset_folder_list, asset_path=GEE_PATHS[app_type]["GEE_ASSET_PATH"] + ) + + description_zoi_ci + ) + delete_asset_on_GEE(asset_id_zoi_ci) if roi: roi = ee.FeatureCollection(roi) else: @@ -56,20 +73,10 @@ def generate_zoi_ci( asset_folder_list=asset_folder_list, asset_suffix=asset_suffix, app_type=app_type, - start_year=2017, - end_year=2024, + start_year=start_year, + end_year=end_year, gee_account_id=gee_account_id, ) - start_date = "2017-07-01" - end_date = "2025-06-30" - description_zoi_ci = f"cropping_intensity_zoi_{asset_suffix}" - - asset_id_zoi_ci = ( - get_gee_dir_path( - asset_folder_list, asset_path=GEE_PATHS[app_type]["GEE_ASSET_PATH"] - ) - + description_zoi_ci - ) if state and district and block: layer_name = f"waterbodies_zoi_{asset_suffix}" layer_at_geoserver = sync_asset_to_db_and_geoserver( diff --git a/computing/zoi_layers/zoi3.py b/computing/zoi_layers/zoi3.py index fea3b962..afecc374 100644 --- a/computing/zoi_layers/zoi3.py +++ b/computing/zoi_layers/zoi3.py @@ -9,7 +9,7 @@ get_gee_dir_path, is_gee_asset_exists, ) -from waterrejuvenation.utils import wait_for_task_completion +from waterrejuvenation.utils import wait_for_task_completion, resolve_zoi_ndvi_input import ee @@ -20,32 +20,25 @@ def get_ndvi_for_zoi( zoi_roi=None, asset_suffix=None, asset_folder_list=None, - start_year="2017-23", + start_date=None, + end_date=None, + start_year=None, + end_year=None, app_type="MWS", gee_account_id=None, proj_id=None, ): print("started generating ndvi") + if not start_date or not end_date or start_year is None or end_year is None: + raise ValueError( + "start_date, end_date, start_year, and end_year are required for ZOI NDVI." + ) ee_initialize(gee_account_id) from waterrejuvenation.utils import get_ndvi_data - if not proj_id: - description_zoi = "cropping_intensity_zoi_" + asset_suffix - asset_id_zoi = ( - get_gee_dir_path( - asset_folder_list, asset_path=GEE_PATHS[app_type]["GEE_ASSET_PATH"] - ) - + description_zoi - ) - else: - - description_zoi = "cropping_intensity_zoi_" + asset_suffix - asset_id_zoi = ( - get_gee_dir_path( - asset_folder_list, asset_path=GEE_PATHS[app_type]["GEE_ASSET_PATH"] - ) - + description_zoi - ) + zoi_collections = resolve_zoi_ndvi_input( + asset_folder_list, app_type, asset_suffix, zoi_roi=zoi_roi + ) description_ndvi = asset_suffix ndvi_asset_path = ( @@ -55,15 +48,14 @@ def get_ndvi_for_zoi( + description_ndvi ) - zoi_collections = ee.FeatureCollection(asset_id_zoi) - fc = get_ndvi_data(zoi_collections, 2017, 2024, description_ndvi, ndvi_asset_path) + fc = get_ndvi_data( + zoi_collections, start_year, end_year, description_ndvi, ndvi_asset_path + ) task = ee.batch.Export.table.toAsset( collection=fc, description=description_ndvi, assetId=ndvi_asset_path ) task.start() wait_for_task_completion(task) - start_date = "30-06-2017" - end_date = "01-07-2024" if state and district and block: layer_name = f"waterbodies_zoi_{asset_suffix}" layer_at_geoserver = sync_asset_to_db_and_geoserver( diff --git a/computing/zoi_layers/zoi_ndvi_from_timeseries.py b/computing/zoi_layers/zoi_ndvi_from_timeseries.py new file mode 100644 index 00000000..3bf3a9d7 --- /dev/null +++ b/computing/zoi_layers/zoi_ndvi_from_timeseries.py @@ -0,0 +1,209 @@ +""" +ZOI NDVI using the ndvi_timeseries compute (HLS-interpolated NDVI). + +Uses the fast band-stack + single reduceRegions path (same idea as ndvi_time_series._generate_ndvi), +while keeping the original NDVI_ JSON property output shape. +""" + +import ee + +from computing.misc.hls_interpolated_ndvi import get_padded_ndvi_ts_image +from computing.surface_water_bodies.swb import sync_asset_to_db_and_geoserver +from computing.utils import sync_project_fc_to_geoserver +from projects.models import Project +from utilities.constants import GEE_PATHS +from utilities.gee_utils import ( + ee_initialize, + get_gee_dir_path, + check_task_status, + is_gee_asset_exists, + export_vector_asset_to_gee, +) +from waterrejuvenation.utils import wait_for_task_completion, resolve_zoi_ndvi_input + + +def _merge_ndvi_year_assets(chunk_assets): + """Same merge logic as waterrejuvenation.utils.merge_assets_chunked_on_year.""" + + def merge_features(feature): + uid = feature.get("UID") + matched_features = [] + for i in range(1, len(chunk_assets)): + matched_feature = ee.Feature( + ee.FeatureCollection(chunk_assets[i]) + .filter(ee.Filter.eq("UID", uid)) + .first() + ) + matched_features.append(matched_feature) + + merged_properties = feature.toDictionary() + for f in matched_features: + merged_properties = merged_properties.combine( + f.toDictionary(), overwrite=False + ) + + return ee.Feature(feature.geometry(), merged_properties) + + return ee.FeatureCollection(chunk_assets[0]).map(merge_features) + + +def _ndvi_bands_to_json_property(feature, year): + """Convert toBands reduceRegions output into NDVI_ JSON string.""" + + props = feature.toDictionary() + keys = props.keys().filter(ee.Filter.stringContains("item", "20")) + + def build_dict(k, acc): + k = ee.String(k) + # toBands prefixes band names with "_" — drop that index + new_key = k.split("_").slice(1).join("_") + return ee.Dictionary(acc).set(new_key, props.get(k)) + + ndvi_dict = ee.Dictionary(keys.iterate(build_dict, ee.Dictionary({}))) + ndvi_json = ee.String.encodeJSON(ndvi_dict) + return feature.set(f"NDVI_{year}", ndvi_json) + + +def build_ndvi_timeseries_from_timeseries_compute( + suitability_vector, + start_year, + end_year, + description, + asset_id, + ndvi_interval_days=16, + reducer_scale=30, + tile_scale=4, +): + """ + Build NDVI time series for ZOI polygons (NDVI_ JSON per feature). + + Fast path: one reduceRegions per hydrological year (not per timestep). + """ + feature_count = suitability_vector.size().getInfo() + print("total") + print(feature_count) + if feature_count == 0: + raise ValueError("Cannot generate NDVI: suitability vector has 0 features") + + task_ids = [] + asset_ids = [] + year = start_year + + while year <= end_year: + start_date = f"{year}-07-01" + end_date = f"{year + 1}-06-30" + ndvi_description = f"ndvi_{year}_{description}" + ndvi_asset_id = f"{asset_id}_ndvi_{year}" + + if is_gee_asset_exists(ndvi_asset_id): + ee.data.deleteAsset(ndvi_asset_id) + + ndvi = get_padded_ndvi_ts_image( + start_date, end_date, suitability_vector.bounds(), ndvi_interval_days + ) + + # Stack all dates into one image, then a single reduceRegions (fast) + ndvi_stack = ndvi.map( + lambda img: img.select("gapfilled_NDVI_lsc").rename( + img.date().format("YYYY-MM-dd") + ) + ).toBands() + + reduced = ndvi_stack.reduceRegions( + collection=suitability_vector, + reducer=ee.Reducer.mean(), + scale=reducer_scale, + tileScale=tile_scale, + ) + + merged_fc = reduced.map(lambda f: _ndvi_bands_to_json_property(f, year)) + + try: + task_id = export_vector_asset_to_gee( + merged_fc, ndvi_description, ndvi_asset_id + ) + if task_id: + print(f"Started export for {year}") + asset_ids.append(ndvi_asset_id) + task_ids.append(task_id) + except Exception as e: + print("Export error:", e) + + year += 1 + + check_task_status(task_ids) + return _merge_ndvi_year_assets(asset_ids) + + +def get_ndvi_for_zoi_from_timeseries_compute( + state=None, + district=None, + block=None, + zoi_roi=None, + asset_suffix=None, + asset_folder_list=None, + start_date=None, + end_date=None, + start_year=None, + end_year=None, + app_type="MWS", + gee_account_id=None, + proj_id=None, +): + """ + Same orchestration and result as get_ndvi_for_zoi (zoi3), but NDVI values come from + the ndvi_timeseries compute (get_padded_ndvi_ts_image) instead of get_ndvi_data. + """ + print("started generating ndvi for zoi (timeseries compute, fast reduce)") + if not start_date or not end_date or start_year is None or end_year is None: + raise ValueError( + "start_date, end_date, start_year, and end_year are required for ZOI NDVI." + ) + ee_initialize(gee_account_id) + + zoi_collections = resolve_zoi_ndvi_input( + asset_folder_list, app_type, asset_suffix, zoi_roi=zoi_roi + ) + + description_ndvi = asset_suffix + ndvi_asset_path = ( + get_gee_dir_path( + asset_folder_list, asset_path=GEE_PATHS[app_type]["GEE_ASSET_PATH"] + ) + + description_ndvi + ) + + fc = build_ndvi_timeseries_from_timeseries_compute( + zoi_collections, + start_year, + end_year, + description_ndvi, + ndvi_asset_path, + ) + + task = ee.batch.Export.table.toAsset( + collection=fc, description=description_ndvi, assetId=ndvi_asset_path + ) + task.start() + wait_for_task_completion(task) + + if state and district and block: + layer_name = f"waterbodies_zoi_{asset_suffix}" + sync_asset_to_db_and_geoserver( + ndvi_asset_path, + layer_name, + asset_suffix, + start_date, + end_date, + state, + district, + block, + ) + elif proj_id: + proj_obj = Project.objects.get(pk=proj_id) + layer_name = f"waterbodies_zoi_{asset_suffix}" + sync_project_fc_to_geoserver( + fc, proj_obj.name, layer_name, "zoi_layers" + ) + + return fc diff --git a/waterrejuvenation/models.py b/waterrejuvenation/models.py index 3f4397fb..b6b4afd1 100644 --- a/waterrejuvenation/models.py +++ b/waterrejuvenation/models.py @@ -62,6 +62,9 @@ def __str__(self): def save(self, *args, **kwargs): """Override save to calculate file hash before saving""" + start_date = kwargs.pop("start_date", None) + end_date = kwargs.pop("end_date", None) + if not self.excel_hash and self.file: # Calculate hash for new file self.file.seek(0) @@ -74,6 +77,11 @@ def save(self, *args, **kwargs): print(f"is processing required: {self.is_processing_required}") print(f"is lullc required: {self.is_lulc_required}") if self.is_compute: + if self.is_processing_required and (not start_date or not end_date): + raise ValueError( + "start_date and end_date are required when is_compute and " + "is_processing_required are true (YYYY-MM-DD)." + ) Upload_Desilting_Points.apply_async( kwargs={ "file_obj_id": self.id, @@ -81,6 +89,8 @@ def save(self, *args, **kwargs): "is_lulc_required": self.is_lulc_required, "is_processing_required": self.is_processing_required, "is_closest_wp": self.is_closest_wp, + "start_date": start_date, + "end_date": end_date, }, queue="waterbody1", ) diff --git a/waterrejuvenation/tasks.py b/waterrejuvenation/tasks.py index 290669fd..4f262062 100644 --- a/waterrejuvenation/tasks.py +++ b/waterrejuvenation/tasks.py @@ -22,7 +22,7 @@ from computing.water_rejuvenation.water_rejuventation import ( find_watersheds_for_point_with_buffer, ) -from computing.zoi_layers.zoi import generate_zoi +from computing.zoi_layers.zoi import generate_zoi, _resolve_zoi_time_window from projects.models import Project from utilities.constants import SITE_DATA_PATH, GEE_PATHS @@ -40,6 +40,8 @@ wait_for_task_completion, delete_asset_on_GEE, find_nearest_water_pixel, + format_waterbody_uid_value, + id_text, ) from computing.surface_water_bodies.swb import generate_swb_layer @@ -64,33 +66,186 @@ def is_nan(value): ) -def _gee_safe_property(value): +def _is_string_id_property(prop_name): + """Identifier / label fields must stay strings in GEE + GeoServer exports.""" + lowered = str(prop_name).lower() + if lowered in { + "uid", + "mws_uid", + "desilt_id", + "pond_id", + "village_id", + "waterbody_name", + "village", + "state", + "district", + "taluka", + "tehsil", + "block", + }: + return True + return lowered.endswith("_uid") or lowered.endswith("_id") + + +def _gee_safe_property(value, prop_name=None): + import json import numpy as np + from datetime import date, datetime + from shapely.geometry.base import BaseGeometry if is_nan(value): - return "N/A" + return None + if prop_name and _is_string_id_property(prop_name): + text = id_text(value) + return text + if isinstance(value, BaseGeometry): + return value.wkt + if isinstance(value, (np.bool_,)): + return bool(value) if isinstance(value, (np.integer,)): return int(value) if isinstance(value, (np.floating,)): return float(value) if isinstance(value, bool): return value + if isinstance(value, int): + return value + if isinstance(value, float): + return value + if isinstance(value, (datetime, date, pd.Timestamp)): + return value.isoformat() + if isinstance(value, (list, tuple, set)): + return json.dumps( + [_gee_safe_property(item) for item in value], + default=str, + ) + if isinstance(value, dict): + # GeoJSON Feature/Geometry dicts cannot be stored as table properties. + if value.get("type") in { + "Feature", + "FeatureCollection", + "Point", + "MultiPoint", + "LineString", + "MultiLineString", + "Polygon", + "MultiPolygon", + "GeometryCollection", + }: + return json.dumps(value) + return json.dumps( + {str(k): _gee_safe_property(v) for k, v in value.items()}, + default=str, + ) + if isinstance(value, str): + stripped = value.strip() + if not stripped or stripped.upper() in ("N/A", "NAN", "NONE"): + return None + if prop_name and prop_name.lower() in {"area_ored", "zoi", "zoi_wb", "zoi_area"}: + try: + return float(stripped) + except ValueError: + return stripped + return stripped return str(value) +def _sync_gdf_to_project_geoserver(gdf, project_name, layer_name, workspace): + """Push a local GeoDataFrame to GeoServer without an EE getInfo round-trip.""" + import os + + import geopandas as gpd + + from computing.utils import fix_invalid_geometry_in_gdf, push_shape_to_geoserver + + if gdf is None or gdf.empty: + logger.warning("No features to sync for layer %s", layer_name) + return None + + state_dir = os.path.join("data/fc_to_shape", project_name) + os.makedirs(state_dir, exist_ok=True) + path = os.path.join(state_dir, layer_name) + out_gdf = gpd.GeoDataFrame(gdf.copy(), geometry="geometry", crs="EPSG:4326") + out_gdf = fix_invalid_geometry_in_gdf(out_gdf) + out_gdf.to_file(path + ".gpkg", driver="GPKG") + return push_shape_to_geoserver( + path, workspace=workspace, layer_name=layer_name, file_type="gpkg" + ) + + def _gdf_to_ee_feature_collection(gdf): - features = [] + gdf = _normalize_uid_in_gdf(gdf) + geojson_features = [] for _, row in gdf.iterrows(): geom = row.geometry if geom is None or geom.is_empty: continue - props = { - col: _gee_safe_property(row[col]) - for col in gdf.columns - if col != "geometry" - } - features.append(ee.Feature(ee.Geometry(geom.__geo_interface__), props)) - return ee.FeatureCollection(features) + props = {} + for col in gdf.columns: + if col == "geometry": + continue + safe_val = _gee_safe_property(row[col], prop_name=col) + if safe_val is not None: + props[col] = safe_val + geojson_features.append( + { + "type": "Feature", + "geometry": geom.__geo_interface__, + "properties": props, + } + ) + if not geojson_features: + return ee.FeatureCollection([]) + return ee.FeatureCollection(geojson_features) + + +def _extract_uid(row): + """Read waterbody UID from a GeoDataFrame row (handles casing / aliases).""" + if row is None: + return None + mws_val = None + uid_val = None + for key in ("MWS_UID", "mws_uid"): + if key in row.index: + mws_val = row[key] + break + for key in ("UID", "uid", "Uid"): + if key in row.index: + uid_val = row[key] + break + uid, _ = format_waterbody_uid_value(mws_val, uid_val) + return uid + + +def _normalize_uid_in_gdf(gdf): + """Ensure canonical UID / MWS_UID columns exist with underscore format.""" + if gdf.empty: + return gdf + gdf = gdf.copy() + for key in ("uid", "Uid", "MWS_UID", "mws_uid"): + if key in gdf.columns and "UID" not in gdf.columns: + gdf["UID"] = gdf[key] + break + if "UID" not in gdf.columns: + return gdf + + mws_col = next( + (col for col in ("MWS_UID", "mws_uid") if col in gdf.columns), + None, + ) + + def _normalize_row(row): + mws_val = row[mws_col] if mws_col else None + uid, mws_val = format_waterbody_uid_value(mws_val, row["UID"]) + row = row.copy() + if uid is not None: + row["UID"] = uid + if mws_val is not None and mws_col: + row[mws_col] = mws_val + return row + + gdf = gdf.apply(_normalize_row, axis=1) + return gdf def _fc_to_gdf(feature_collection): @@ -100,7 +255,57 @@ def _fc_to_gdf(feature_collection): return gpd.GeoDataFrame( columns=["geometry"], geometry="geometry", crs="EPSG:4326" ) - return gpd.GeoDataFrame.from_features(features, crs="EPSG:4326") + gdf = gpd.GeoDataFrame.from_features(features, crs="EPSG:4326") + return _normalize_uid_in_gdf(gdf) + + +def _swb_polygon_geometry(wb_row): + """Return SWB polygon geometry for matched water-rejuvenation exports.""" + geom = wb_row.geometry + if geom is None or geom.is_empty: + return None + if geom.geom_type in ("Polygon", "MultiPolygon"): + return geom + logger.warning( + "SWB feature %s has geometry type %s; expected Polygon/MultiPolygon.", + wb_row.name, + geom.geom_type, + ) + return geom + + +def _merge_matched_desilt_with_swb(desilt_row, wb_row): + """Matched export: SWB polygon geometry + SWB and desilting properties.""" + geometry = _swb_polygon_geometry(wb_row) + if geometry is None: + return None + + wb_props = { + col: wb_row[col] + for col in wb_row.index + if col != "geometry" + } + desilt_props = { + col: desilt_row[col] + for col in desilt_row.index + if col != "geometry" and col.lower() not in ("uid", "mws_uid") + } + props = {**wb_props, **desilt_props, "matched": True} + mws_val = wb_row["MWS_UID"] if "MWS_UID" in wb_row.index else None + if mws_val is None and "mws_uid" in wb_row.index: + mws_val = wb_row["mws_uid"] + uid_val = wb_row["UID"] if "UID" in wb_row.index else None + uid, mws_val = format_waterbody_uid_value(mws_val, uid_val) + if mws_val is not None: + props["MWS_UID"] = mws_val + if uid: + props["UID"] = uid + elif "UID" not in props: + logger.warning( + "Matched waterbody has no UID (wb_index=%s); ZOI merge may fail.", + wb_row.name, + ) + return {**props, "geometry": geometry} def _match_desilting_points_to_waterbodies(desilt_gdf, wb_gdf, max_distance_m=100): @@ -122,28 +327,29 @@ def _match_desilting_points_to_waterbodies(desilt_gdf, wb_gdf, max_distance_m=10 for idx, point_row in desilt_gdf.iterrows(): point_metric = desilt_metric.loc[idx].geometry - props = {col: point_row[col] for col in desilt_gdf.columns if col != "geometry"} + desilt_props = { + col: point_row[col] for col in desilt_gdf.columns if col != "geometry" + } intersect_hits = wb_metric[wb_metric.intersects(point_metric)] if not intersect_hits.empty: - wb_idx = intersect_hits.index[0] - out_geom = wb_gdf.loc[wb_idx].geometry - props["matched"] = True - props["match_type"] = "intersect" - matched_rows.append({**props, "geometry": out_geom}) + wb_row = wb_gdf.loc[intersect_hits.index[0]] + row = _merge_matched_desilt_with_swb(point_row, wb_row) + if row: + row["match_type"] = "intersect" + matched_rows.append(row) continue near_hits = wb_metric[wb_metric.intersects(point_metric.buffer(max_distance_m))] if not near_hits.empty: - wb_idx = near_hits.index[0] - out_geom = wb_gdf.loc[wb_idx].geometry - props["matched"] = True - props["match_type"] = "near" - matched_rows.append({**props, "geometry": out_geom}) + wb_row = wb_gdf.loc[near_hits.index[0]] + row = _merge_matched_desilt_with_swb(point_row, wb_row) + if row: + row["match_type"] = "near" + matched_rows.append(row) continue - props["matched"] = False - props["match_type"] = "none" + props = {**desilt_props, "matched": False, "match_type": "none"} unmatched_rows.append({**props, "geometry": point_row.geometry}) matched_gdf = ( @@ -170,6 +376,8 @@ def Upload_Desilting_Points( is_lulc_required=True, gee_account_id=None, is_processing_required=True, + start_date=None, + end_date=None, ): import pandas as pd from .models import WaterbodiesFileUploadLog, WaterbodiesDesiltingLog @@ -181,6 +389,16 @@ def normalize(val): return None return val + if (is_processing_required or is_closest_wp) and (not start_date or not end_date): + raise ValueError( + "start_date and end_date are required when processing desilting points " + "or finding the closest water pixel (YYYY-MM-DD)." + ) + + start_year = end_year = None + if start_date and end_date: + _, _, start_year, end_year = _resolve_zoi_time_window(start_date, end_date) + ee_initialize(gee_account_id) wb_obj = WaterbodiesFileUploadLog.objects.get(pk=file_obj_id) @@ -230,7 +448,11 @@ def normalize(val): print("inside closest wp") try: result_dict = find_nearest_water_pixel( - dsilting_obj_log.lat, dsilting_obj_log.lon, 1500 + dsilting_obj_log.lat, + dsilting_obj_log.lon, + 1500, + start_year=start_year, + end_year=end_year, ) print(result_dict) except Exception as e: @@ -298,6 +520,8 @@ def normalize(val): is_lulc_required=is_lulc_required, gee_account_id=gee_account_id, proj_id=proj_obj.id, + start_date=start_date, + end_date=end_date, ) wb_obj.process = True @@ -309,7 +533,12 @@ def Generate_lulc_mws( is_lulc_required=True, gee_account_id=None, proj_id=None, + start_date=None, + end_date=None, ): + start_date, end_date, start_year, end_year = _resolve_zoi_time_window( + start_date, end_date + ) proj_obj = Project.objects.get(pk=proj_id) asset_suffix = f"{proj_obj.name}_{proj_obj.id}".lower() asset_folder = [proj_obj.name.lower()] @@ -335,8 +564,8 @@ def Generate_lulc_mws( make_asset_public(mws_asset_id) if is_lulc_required: clip_lulc_v3( - start_year=2017, - end_year=2024, + start_year=start_year, + end_year=end_year, gee_account_id=gee_account_id, roi_path=mws_asset_id, asset_folder=asset_folder, @@ -347,7 +576,11 @@ def Generate_lulc_mws( except Exception as e: logger.error(f"Error in Generating Lulc and mws layer: {str(e)}") Generate_water_balance_indicator( - mws_asset_id, proj_id=proj_obj.id, gee_account_id=gee_account_id + mws_asset_id, + proj_id=proj_obj.id, + gee_account_id=gee_account_id, + start_date=start_date, + end_date=end_date, ) asset_suffix_swb3 = f"swb3_{proj_obj.name}+{proj_obj.id}" asset_id_swb = ( @@ -357,7 +590,10 @@ def Generate_lulc_mws( + asset_suffix_swb3 ) BuildMWSLayer( - gee_account_id=gee_account_id, proj_id=proj_obj.id, app_type="WATERBODY" + gee_account_id=gee_account_id, + proj_id=proj_obj.id, + app_type="WATERBODY", + export_year_range=(start_year, end_year), ) asset_suffix_wb = f"waterbodies_{asset_suffix}".lower() asset_id_wb = ( @@ -373,13 +609,20 @@ def Generate_lulc_mws( asset_suffix=asset_suffix, asset_folder=asset_folder, app_type="WATERBODY", + start_date=start_date, + end_date=end_date, ) @shared_task() -def Generate_water_balance_indicator(mws_asset_id, proj_id, gee_account_id=None): +def Generate_water_balance_indicator( + mws_asset_id, proj_id, gee_account_id=None, start_date=None, end_date=None +): print(f"project id {gee_account_id}") + start_date, end_date, start_year, end_year = _resolve_zoi_time_window( + start_date, end_date + ) proj_obj = Project.objects.get(pk=proj_id) logger.info("Generating SWB layer for given lat long") asset_folder = [str(proj_obj.name).lower()] @@ -430,8 +673,8 @@ def Generate_water_balance_indicator(mws_asset_id, proj_id, gee_account_id=None) asset_suffix=asset_suffix, asset_folder_list=asset_folder, app_type="WATERBODY", - start_year="2017", - end_year="2024", + start_year=str(start_year), + end_year=str(end_year), is_all_classes=True, gee_account_id=gee_account_id, ) @@ -455,8 +698,8 @@ def Generate_water_balance_indicator(mws_asset_id, proj_id, gee_account_id=None) asset_suffix=asset_suffix, asset_folder_list=asset_folder, app_type="WATERBODY", - start_date="2017-06-30", - end_date="2025-07-1", + start_date=start_date, + end_date=end_date, is_annual=False, ) make_asset_public(asset_id_prec) @@ -466,12 +709,14 @@ def Generate_water_balance_indicator(mws_asset_id, proj_id, gee_account_id=None) asset_suffix=asset_suffix, asset_folder_list=asset_folder, app_type="WATERBODY", - start_year=2017, - end_year=2024, + start_year=start_year, + end_year=end_year, gee_account_id=gee_account_id, state=proj_obj.state_soi.state_name, ) - dst_filename = "drought_" + asset_suffix + "_" + str(2017) + "_" + str(2022) + dst_filename = ( + "drought_" + asset_suffix + "_" + str(start_year) + "_" + str(end_year) + ) draught_asset_id = ( get_gee_dir_path( asset_folder, asset_path=GEE_PATHS["WATERBODY"]["GEE_ASSET_PATH"] @@ -513,6 +758,8 @@ def Genereate_zoi_and_zoi_indicator( asset_suffix=None, asset_folder=None, roi=None, + start_date=None, + end_date=None, ): print(f"roi: {roi}") ee_initialize(gee_project_id) @@ -531,6 +778,8 @@ def Genereate_zoi_and_zoi_indicator( app_type=app_type, gee_account_id=gee_project_id, proj_id=proj_id, + start_date=start_date, + end_date=end_date, ) @@ -655,7 +904,7 @@ def BuildMWSLayer( block=None, district=None, drought_asset_override=None, # optional: full path to drought asset if you want to override default - export_year_range=(2017, 2022), # for naming drought asset + export_year_range=None, # (start_year, end_year) for naming drought asset ): """ Full BuildMWSLayer: builds final MWS waterbody FC, joins drought properties (flat, prefixed), @@ -673,6 +922,10 @@ def BuildMWSLayer( try: # initialize GEE + if export_year_range is None: + raise ValueError( + "export_year_range=(start_year, end_year) is required for BuildMWSLayer." + ) ee_initialize(gee_account_id) # ------------------------- @@ -724,10 +977,10 @@ def BuildMWSLayer( # ------------------------- # Drought asset id (default naming) # ------------------------- + start_y, end_y = export_year_range if drought_asset_override: draught_asset_id = drought_asset_override else: - start_y, end_y = export_year_range dst_filename = f"drought_{asset_suffix}" draught_asset_id = ( get_gee_dir_path( @@ -755,8 +1008,12 @@ def BuildMWSLayer( # ------------------------- # Create final_fc using your domain function # ------------------------- + start_y, end_y = export_year_range final_fc = calculate_precipitation_season( - mws_geojson_op, draught_asset_id=draught_asset_id + mws_geojson_op, + draught_asset_id=draught_asset_id, + start_year=start_y, + end_year=end_y, ) final_fc = ee.FeatureCollection(final_fc) @@ -983,6 +1240,10 @@ def BuildWaterBodyLayer( matched_gdf, unmatched_gdf = _match_desilting_points_to_waterbodies( desilt_gdf, wb_gdf, max_distance_m=100 ) + matched_gdf = _normalize_uid_in_gdf(matched_gdf) + if len(matched_gdf) > 0: + geom_types = matched_gdf.geometry.geom_type.value_counts().to_dict() + logger.info("BuildWaterBodyLayer matched geometry types: %s", geom_types) matched_fc = _gdf_to_ee_feature_collection(matched_gdf) unmatched_fc = _gdf_to_ee_feature_collection(unmatched_gdf) @@ -1048,8 +1309,8 @@ def BuildWaterBodyLayer( # ------------------------------------------------------------------ layer_name = f"waterbodies_{proj_obj.name}_{proj_obj.id}".lower() if len(matched_gdf) > 0: - sync_project_fc_to_geoserver( - matched_fc, + _sync_gdf_to_project_geoserver( + matched_gdf, proj_obj.name, layer_name, "swb", diff --git a/waterrejuvenation/utils.py b/waterrejuvenation/utils.py index b61fb7b7..667d318a 100644 --- a/waterrejuvenation/utils.py +++ b/waterrejuvenation/utils.py @@ -30,7 +30,7 @@ logger = logging.getLogger(__name__) from datetime import datetime -from nrm_app.settings import lulc_years, water_classes +from nrm_app.settings import water_classes import pandas as pd import math @@ -38,16 +38,6 @@ import requests import json -years = [ - "2017_2018", - "2018_2019", - "2019_2020", - "2020_2021", - "2021_2022", - "2022_2023", - "2023_2024", -] - # utils.py EXPECTED_EXCEL_HEADERS = { @@ -160,16 +150,16 @@ def fine_closest_wb_pixel(lon, lat): ee_initialize() -def find_nearest_water_pixel(lat, lon, distance_threshold): +def find_nearest_water_pixel(lat, lon, distance_threshold, start_year=None, end_year=None): """ Finds the nearest water pixel within a given threshold. Parameters: lat (float): Latitude of the input location. lon (float): Longitude of the input location. - lulc_years (list of str): List of LULC year identifiers (e.g., '2017_2018'). - water_classes (list of int): List of LULC class codes considered as water. distance_threshold (float): Maximum distance (in meters) to consider. + start_year (int): First hydrological year start (required). + end_year (int): Last hydrological year start (required). Returns: dict: { @@ -179,6 +169,16 @@ def find_nearest_water_pixel(lat, lon, distance_threshold): 'distance_m': float (if success) } """ + if start_year is None or end_year is None: + raise ValueError( + "start_year and end_year are required for find_nearest_water_pixel." + ) + start_year = int(start_year) + end_year = int(end_year) + if start_year > end_year: + raise ValueError("start_year must be less than or equal to end_year.") + + lulc_year_ids = [f"{y}_{y + 1}" for y in range(start_year, end_year + 1)] print("Given lat long") print(f"-----{lat}----{lon}---") @@ -186,7 +186,7 @@ def find_nearest_water_pixel(lat, lon, distance_threshold): # Create water masks from each LULC year water_masks = [] - for year in lulc_years: + for year in lulc_year_ids: image = ee.Image(f"{PAN_INDIA_LULC_V3_DATASET}{year}") water_mask = ( image.select("predicted_label") @@ -340,7 +340,15 @@ def create_ring(feature): zoi = ee.Number(feature.get("zoi_wb")) waterbody_name = feature.get("waterbody_name") impactfull = feature.get("impactful") - uid = feature.get("UID") + uid = ee.Algorithms.If( + feature.get("UID"), + feature.get("UID"), + ee.Algorithms.If( + feature.get("uid"), + feature.get("uid"), + feature.get("MWS_UID"), + ), + ) # Make circle buffer from centroid centroid = geom.centroid() @@ -403,6 +411,131 @@ def calculate_zoi_area(zoi): return ee.Number.parse(area_hectares.format("%.2f")) +def is_nan_value(value): + return ( + value is None + or (isinstance(value, float) and math.isnan(value)) + or pd.isna(value) + ) + + +def id_text(value): + """Normalize an identifier value to a clean string.""" + if is_nan_value(value): + return None + if isinstance(value, float) and value.is_integer(): + value = int(value) + if isinstance(value, int): + text = str(value) + else: + text = str(value).strip() + if not text or text.upper() in ("N/A", "NAN", "NONE"): + return None + if text.endswith(".0"): + stem = text[:-2] + if stem.replace("_", "").isdigit(): + text = stem + if "e" in text.lower(): + try: + as_float = float(text) + if as_float.is_integer(): + text = str(int(as_float)) + except ValueError: + pass + return text + + +def format_waterbody_uid_value(mws_uid, uid): + """ + Restore SWB UID format: _, e.g. 25_34833_101. + + GEE/pandas may coerce this to a number (12129065580), dropping the underscore + before the per-waterbody index (12129065_580). + """ + mws_str = id_text(mws_uid) + uid_str = id_text(uid) + if not uid_str: + return None, mws_str + if "_" in uid_str: + return uid_str, mws_str + if not mws_str: + return uid_str, mws_str + + mws_digits = mws_str.replace("_", "") + uid_digits = uid_str.replace("_", "") + if uid_digits.startswith(mws_digits) and len(uid_digits) > len(mws_digits): + suffix = uid_digits[len(mws_digits) :] + if suffix.isdigit(): + return f"{mws_str}_{suffix}", mws_str + return uid_str, mws_str + + +def ensure_uid_on_fc(fc): + """Ensure every feature has a canonical UID for NDVI / merge steps.""" + + def ensure_uid(feature): + uid = ee.Algorithms.If( + feature.get("UID"), + feature.get("UID"), + ee.Algorithms.If( + feature.get("uid"), + feature.get("uid"), + ee.Algorithms.If( + feature.get("MWS_UID"), + feature.get("MWS_UID"), + feature.get("system:index"), + ), + ), + ) + return feature.set("UID", uid) + + return fc.map(ensure_uid) + + +def resolve_zoi_ndvi_input( + asset_folder_list, app_type, asset_suffix, zoi_roi=None +): + """ + Pick polygons for ZOI NDVI: prefer cropping_intensity_zoi, else zoi_ rings. + """ + if zoi_roi: + fc = ee.FeatureCollection(zoi_roi) + count = fc.size().getInfo() + logger.info("NDVI input: explicit roi (%s features)", count) + if count == 0: + raise ValueError(f"NDVI input roi is empty: {zoi_roi}") + return ensure_uid_on_fc(fc) + + base = get_gee_dir_path( + asset_folder_list, asset_path=GEE_PATHS[app_type]["GEE_ASSET_PATH"] + ) + ci_asset = base + f"cropping_intensity_zoi_{asset_suffix}" + zoi_asset = base + f"zoi_{asset_suffix}" + + ci_fc = ee.FeatureCollection(ci_asset) + ci_size = ci_fc.size().getInfo() + if ci_size > 0: + logger.info( + "NDVI input: cropping_intensity_zoi (%s features) from %s", + ci_size, + ci_asset, + ) + return ensure_uid_on_fc(ci_fc) + + zoi_fc = ee.FeatureCollection(zoi_asset) + zoi_size = zoi_fc.size().getInfo() + logger.info( + "NDVI input: fallback zoi_ (%s features) from %s", + zoi_size, + zoi_asset, + ) + if zoi_size == 0: + raise ValueError( + f"No ZOI features for NDVI (tried {ci_asset} and {zoi_asset})" + ) + return ensure_uid_on_fc(zoi_fc) + + def get_ndvi_data(suitability_vector, start_year, end_year, description, asset_id): """ Extracts and exports NDVI data for a set of features by aggregating NDVI values @@ -421,6 +554,13 @@ def get_ndvi_data(suitability_vector, start_year, end_year, description, asset_i Returns: ee.FeatureCollection: Merged NDVI time series across years. """ + suitability_vector = ensure_uid_on_fc(suitability_vector) + feature_count = suitability_vector.size().getInfo() + print("total") + print(feature_count) + if feature_count == 0: + raise ValueError("Cannot generate NDVI: suitability vector has 0 features") + task_ids = [] asset_ids = [] # Loop over each year @@ -465,39 +605,18 @@ def annotate(feature): # Map image-wise extraction and flatten to a single FeatureCollection all_ndvi = ndvi.map(map_image).flatten() - # Extract all unique UIDs from the input feature collection - uids = suitability_vector.aggregate_array("UID") - count = uids.size() # Server-side count - print("total") - print(count.getInfo()) - - # For each UID, filter NDVI features and aggregate to dict - def build_feature(uid): - """ - Reconstruct a single feature by merging its NDVI values across all images - into one property NDVI_ as a JSON dictionary {date: value}. - """ - # Get the geometry and properties of the original feature - feature_geom = ee.Feature( - suitability_vector.filter(ee.Filter.eq("UID", uid)).first() - ) - - # Filter all NDVI records related to this UID + # Reconstruct one row per input polygon with NDVI_ JSON property + def build_feature(input_feature): + uid = input_feature.get("UID") filtered = all_ndvi.filter(ee.Filter.eq("UID", uid)) - - # Create dictionary: {date: ndvi} date_ndvi_list = filtered.aggregate_array("ndvi_date").zip( filtered.aggregate_array("ndvi") ) - - # Convert to dictionary and encode as JSON string ndvi_dict = ee.Dictionary(date_ndvi_list.flatten()) ndvi_json = ee.String.encodeJSON(ndvi_dict) + return input_feature.set(f"NDVI_{start_year}", ndvi_json) - return feature_geom.set(f"NDVI_{start_year}", ndvi_json) - - # Apply feature-wise aggregation - merged_fc = ee.FeatureCollection(uids.map(build_feature)) + merged_fc = suitability_vector.map(build_feature) # Export as single-row-per-feature collection try: @@ -551,8 +670,18 @@ def merge_features(feature): def get_ndvi_for_zoi( - zoi_asset_path, asset_suffix, asset_folder, proj_id=None, app_type="WATER_REJ" + zoi_asset_path, + asset_suffix, + asset_folder, + proj_id=None, + app_type="WATER_REJ", + start_year=None, + end_year=None, ): + if start_year is None or end_year is None: + raise ValueError( + "start_year and end_year are required for get_ndvi_for_zoi." + ) proj_obj = Project.objects.get(pk=proj_id) asset_suffix_ndvi = f"zoi_ndvi_{proj_obj.name}_{proj_obj.id}" ndvi_asset_path = ( @@ -562,9 +691,13 @@ def get_ndvi_for_zoi( + asset_suffix_ndvi ) - zoi_collections = ee.FeatureCollection(zoi_asset_path) + zoi_collections = resolve_zoi_ndvi_input( + asset_folder, app_type, f"{proj_obj.name}_{proj_obj.id}".lower(), zoi_roi=zoi_asset_path + ) - fc = get_ndvi_data(zoi_collections, 2017, 2024, asset_suffix_ndvi, ndvi_asset_path) + fc = get_ndvi_data( + zoi_collections, start_year, end_year, asset_suffix_ndvi, ndvi_asset_path + ) task = ee.batch.Export.table.toAsset( collection=fc, description=asset_suffix_ndvi, assetId=ndvi_asset_path ) @@ -607,10 +740,17 @@ def merge_features(feature): def _waterbody_area_ha(feature): - """Area in hectares; use geometry when area_ored is missing (e.g. water rej layers).""" + """Area in hectares; use geometry when area_ored is missing or non-numeric.""" geom_area_ha = ee.Number(feature.geometry().area(maxError=1)).divide(10000) - stored_area = feature.get("area_ored") - return ee.Number(ee.Algorithms.If(stored_area, stored_area, geom_area_ha)) + raw_area = feature.get("area_ored") + # Water-rej exports may store area_ored as a string; coerce before numeric ops. + stored_str = ee.String(ee.Algorithms.If(raw_area, raw_area, "0")) + is_na = stored_str.compareTo("N/A").eq(0) + safe_str = ee.String(ee.Algorithms.If(is_na, "0", stored_str)) + stored_numeric = ee.Number.parse(safe_str) + return ee.Number( + ee.Algorithms.If(stored_numeric.gt(0), stored_numeric, geom_area_ha) + ) def compute_zoi(feature): diff --git a/waterrejuvenation/views.py b/waterrejuvenation/views.py index aed9559d..90dcc232 100644 --- a/waterrejuvenation/views.py +++ b/waterrejuvenation/views.py @@ -113,17 +113,36 @@ def create(self, request, *args, **kwargs): continue # Prepare data for serializer + is_compute = request.data.get("is_compute", False) + is_processing_required = request.data.get("is_processing_required", True) + start_date = request.data.get("start_date") or request.data.get("startDate") + end_date = request.data.get("end_date") or request.data.get("endDate") + + def _as_bool(val, default=False): + if val is None: + return default + if isinstance(val, bool): + return val + return str(val).lower() in ("true", "1", "yes") + + compute_enabled = _as_bool(is_compute, False) + processing_enabled = _as_bool(is_processing_required, True) + if compute_enabled and processing_enabled and (not start_date or not end_date): + errors.append( + "start_date and end_date are required when compute processing " + "is enabled (YYYY-MM-DD)." + ) + continue + data = { "name": request.data.get("name", valid_gee_text(uploaded_file.name)), "file": uploaded_file, "project": project.id, "gee_account_id": request.data.get("gee_account_id"), "is_lulc_required": request.data.get("is_lulc_required", True), - "is_processing_required": request.data.get( - "is_processing_required", True - ), + "is_processing_required": is_processing_required, "is_closest_wp": request.data.get("is_closest_wp", True), - "is_compute": request.data.get("is_compute", False), + "is_compute": is_compute, } print(data) serializer = self.get_serializer(data=data) @@ -150,6 +169,8 @@ def create(self, request, *args, **kwargs): project=project, uploaded_by=request.user, excel_hash=excel_hash, + start_date=start_date, + end_date=end_date, ) # # Convert KML to GeoJSON