import uuid from django.db import models def build_default_sub_block(block_code: str, *, boundary: dict | None = None) -> dict: normalized_block_code = str(block_code or "block-1").strip() or "block-1" return { "sub_block_code": f"{normalized_block_code}-sub-1", "cluster_label": 0, "source": "default", "boundary": boundary or {}, "cluster_uuid": None, } def ensure_block_layout_defaults(layout: dict | None, *, block_count: int | None = None) -> dict: raw_layout = dict(layout or {}) raw_blocks = list(raw_layout.get("blocks") or []) normalized_count = len(raw_blocks) if raw_blocks else max(int(block_count or raw_layout.get("input_block_count") or 1), 1) normalized_blocks: list[dict] = [] for index in range(normalized_count): raw_block = raw_blocks[index] if index < len(raw_blocks) else {} block_code = str(raw_block.get("block_code") or f"block-{index + 1}").strip() or f"block-{index + 1}" boundary = raw_block.get("boundary") or {} sub_blocks = [dict(sub_block) for sub_block in (raw_block.get("sub_blocks") or []) if isinstance(sub_block, dict)] if not sub_blocks: sub_blocks = [build_default_sub_block(block_code, boundary=boundary)] normalized_block = { "block_code": block_code, "order": int(raw_block.get("order") or index + 1), "source": raw_block.get("source") or ("input" if raw_blocks or normalized_count > 1 else "default"), "boundary": boundary, "needs_subdivision": raw_block.get("needs_subdivision"), "sub_blocks": sub_blocks, } for extra_key in ("subdivision_summary", "analysis_grid_summary", "aggregated_metrics"): if extra_key in raw_block: normalized_block[extra_key] = raw_block[extra_key] normalized_blocks.append(normalized_block) return { "input_block_count": normalized_count, "default_full_farm": raw_layout.get("default_full_farm", normalized_count == 1), "algorithm_status": raw_layout.get("algorithm_status") or "pending", "blocks": normalized_blocks, } def build_block_layout(block_count: int = 1, blocks: list[dict] | None = None) -> dict: if blocks: return ensure_block_layout_defaults( { "input_block_count": len(blocks), "default_full_farm": len(blocks) == 1, "algorithm_status": "pending", "blocks": [ { "block_code": str(block.get("block_code") or f"block-{index + 1}").strip(), "order": int(block.get("order") or index + 1), "source": "input", "boundary": block.get("boundary") or {}, "needs_subdivision": None, "sub_blocks": [dict(sub_block) for sub_block in (block.get("sub_blocks") or [])], } for index, block in enumerate(blocks) ], }, block_count=len(blocks), ) normalized_count = max(int(block_count or 1), 1) return ensure_block_layout_defaults( { "input_block_count": normalized_count, "default_full_farm": normalized_count == 1, "algorithm_status": "pending", "blocks": [ { "block_code": f"block-{index + 1}", "order": index + 1, "source": "input" if normalized_count > 1 else "default", "boundary": {}, "needs_subdivision": None, "sub_blocks": [], } for index in range(normalized_count) ], }, block_count=normalized_count, ) class SoilLocation(models.Model): """ مرکز زمین و مرز مزرعه/بلوک‌های تعریف‌شده توسط کشاورز. """ latitude = models.DecimalField( max_digits=9, decimal_places=6, db_index=True, help_text="عرض جغرافیایی مرکز زمین (lat)", ) longitude = models.DecimalField( max_digits=9, decimal_places=6, db_index=True, help_text="طول جغرافیایی مرکز زمین (lon)", ) task_id = models.CharField( max_length=255, blank=True, help_text="شناسه تسک Celery در حال پردازش", ) farm_boundary = models.JSONField( default=dict, blank=True, help_text=( "مرز مزرعه برای درخواست‌های سنجش‌ازدور. " 'می‌تواند GeoJSON polygon یا bbox مثل {"type": "Polygon", "coordinates": [...]} باشد.' ), ) input_block_count = models.PositiveIntegerField( default=1, help_text="تعداد بلوک‌های اولیه‌ای که کشاورز برای زمین ثبت می‌کند.", ) block_layout = models.JSONField( default=build_block_layout, blank=True, help_text=( "ساختار بلوک‌های زمین. به‌صورت پیش‌فرض کل زمین یک بلوک است و " "بعداً الگوریتم می‌تواند برای هر بلوک زیر‌بلوک تعریف کند." ), ) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: constraints = [ models.UniqueConstraint( fields=["latitude", "longitude"], name="soil_location_unique_lat_lon", ) ] ordering = ["-updated_at"] verbose_name = "مرکز زمین" verbose_name_plural = "مراکز زمین" def __str__(self): return f"SoilLocation({self.latitude}, {self.longitude})" @property def center_latitude(self): return self.latitude @property def center_longitude(self): return self.longitude @property def is_complete(self): """آیا حداقل یک run کامل remote sensing برای این location وجود دارد؟""" return self.remote_sensing_runs.filter(status="success").exists() def set_input_block_count(self, block_count: int = 1, blocks: list[dict] | None = None): normalized_count = len(blocks) if blocks else max(int(block_count or 1), 1) self.input_block_count = normalized_count self.block_layout = build_block_layout(normalized_count, blocks=blocks) def save(self, *args, **kwargs): if not self.input_block_count: self.input_block_count = 1 self.block_layout = ensure_block_layout_defaults( self.block_layout, block_count=self.input_block_count, ) super().save(*args, **kwargs) class BlockSubdivision(models.Model): """ نتیجه خردسازی یک بلوک برای یک SoilLocation. grid_points نقاط اولیه شبکه هستند و centroid_points مراکز نهایی بخش‌ها. """ soil_location = models.ForeignKey( SoilLocation, on_delete=models.CASCADE, related_name="block_subdivisions", ) block_code = models.CharField( max_length=64, help_text="شناسه بلوکی که این خردسازی برای آن انجام شده است.", ) source_boundary = models.JSONField( default=dict, blank=True, help_text="مرز همان بلوکی که به سرویس subdivision داده شده است.", ) chunk_size_sqm = models.PositiveIntegerField( default=900, help_text="اندازه هر chunk به متر مربع.", ) grid_points = models.JSONField( default=list, blank=True, help_text="نقاط اولیه شبکه داخل مرز بلوک.", ) centroid_points = models.JSONField( default=list, blank=True, help_text="مراکز نهایی بخش‌های خردشده.", ) grid_point_count = models.PositiveIntegerField(default=0) centroid_count = models.PositiveIntegerField(default=0) elbow_plot = models.ImageField( upload_to="location_data/elbow_plots/", null=True, blank=True, help_text="تصویر نمودار elbow برای انتخاب تعداد بهینه خوشه‌ها.", ) status = models.CharField( max_length=32, default="created", help_text="وضعیت تولید subdivision برای این بلوک.", ) metadata = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: constraints = [ models.UniqueConstraint( fields=["soil_location", "block_code"], name="location_block_subdivision_unique_location_block_code", ) ] ordering = ["soil_location", "block_code", "-updated_at"] verbose_name = "خردسازی بلوک" verbose_name_plural = "خردسازی بلوک‌ها" def __str__(self): return f"BlockSubdivision({self.soil_location_id}, {self.block_code})" class RemoteSensingRun(models.Model): STATUS_PENDING = "pending" STATUS_RUNNING = "running" STATUS_SUCCESS = "success" STATUS_FAILURE = "failure" STATUS_CHOICES = [ (STATUS_PENDING, "Pending"), (STATUS_RUNNING, "Running"), (STATUS_SUCCESS, "Success"), (STATUS_FAILURE, "Failure"), ] soil_location = models.ForeignKey( SoilLocation, on_delete=models.CASCADE, related_name="remote_sensing_runs", ) block_subdivision = models.ForeignKey( BlockSubdivision, on_delete=models.SET_NULL, null=True, blank=True, related_name="remote_sensing_runs", ) block_code = models.CharField( max_length=64, blank=True, default="", db_index=True, help_text="شناسه بلوکی که این run برای آن اجرا شده است.", ) provider = models.CharField( max_length=64, default="openeo", help_text="ارائه‌دهنده داده سنجش‌ازدور.", ) chunk_size_sqm = models.PositiveIntegerField( default=900, help_text="اندازه هر سلول تحلیل به متر مربع.", ) temporal_start = models.DateField(null=True, blank=True) temporal_end = models.DateField(null=True, blank=True) status = models.CharField( max_length=16, choices=STATUS_CHOICES, default=STATUS_PENDING, db_index=True, ) metadata = models.JSONField(default=dict, blank=True) error_message = models.TextField(blank=True, default="") started_at = models.DateTimeField(null=True, blank=True) finished_at = models.DateTimeField(null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-created_at", "-id"] indexes = [ models.Index( fields=["soil_location", "status", "created_at"], name="rs_run_loc_status_created_idx", ), models.Index( fields=["block_code", "created_at"], name="rs_run_block_created_idx", ), ] verbose_name = "remote sensing run" verbose_name_plural = "remote sensing runs" def __str__(self): block_text = self.block_code or "farm" return f"RemoteSensingRun({self.soil_location_id}, {block_text}, {self.status})" @property def normalized_status(self) -> str: """ Return the client-facing lifecycle status while keeping legacy DB values stable. """ if self.status == self.STATUS_SUCCESS: return "completed" if self.status == self.STATUS_FAILURE: return "failed" return self.status class AnalysisGridCell(models.Model): soil_location = models.ForeignKey( SoilLocation, on_delete=models.CASCADE, related_name="analysis_grid_cells", ) block_subdivision = models.ForeignKey( BlockSubdivision, on_delete=models.SET_NULL, null=True, blank=True, related_name="analysis_grid_cells", ) block_code = models.CharField( max_length=64, blank=True, default="", db_index=True, help_text="شناسه بلوکی که این سلول به آن تعلق دارد.", ) cell_code = models.CharField( max_length=128, unique=True, help_text="شناسه یکتای سلول تحلیل.", ) chunk_size_sqm = models.PositiveIntegerField( default=900, db_index=True, help_text="اندازه سلول تحلیل به متر مربع.", ) geometry = models.JSONField( default=dict, blank=True, help_text="هندسه سلول به صورت GeoJSON polygon یا ساختار مشابه.", ) centroid_lat = models.DecimalField( max_digits=9, decimal_places=6, db_index=True, help_text="عرض جغرافیایی مرکز سلول.", ) centroid_lon = models.DecimalField( max_digits=9, decimal_places=6, db_index=True, help_text="طول جغرافیایی مرکز سلول.", ) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["soil_location", "block_code", "cell_code"] indexes = [ models.Index( fields=["soil_location", "block_code"], name="grid_cell_loc_block_idx", ), models.Index( fields=["soil_location", "chunk_size_sqm"], name="grid_cell_loc_chunk_idx", ), ] verbose_name = "analysis grid cell" verbose_name_plural = "analysis grid cells" def __str__(self): return f"AnalysisGridCell({self.cell_code})" class AnalysisGridObservation(models.Model): cell = models.ForeignKey( AnalysisGridCell, on_delete=models.CASCADE, related_name="observations", ) run = models.ForeignKey( RemoteSensingRun, on_delete=models.SET_NULL, null=True, blank=True, related_name="observations", ) temporal_start = models.DateField(db_index=True) temporal_end = models.DateField(db_index=True) ndvi = models.FloatField(null=True, blank=True) ndwi = models.FloatField(null=True, blank=True) soil_vv = models.FloatField(null=True, blank=True) soil_vv_db = models.FloatField(null=True, blank=True) dem_m = models.FloatField(null=True, blank=True) slope_deg = models.FloatField(null=True, blank=True) metadata = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-temporal_start", "-temporal_end", "-id"] constraints = [ models.UniqueConstraint( fields=["cell", "temporal_start", "temporal_end"], name="grid_obs_unique_cell_temporal_range", ) ] indexes = [ models.Index( fields=["cell", "temporal_start", "temporal_end"], name="grid_obs_cell_temporal_idx", ), models.Index( fields=["temporal_start", "temporal_end"], name="grid_obs_temporal_idx", ), ] verbose_name = "analysis grid observation" verbose_name_plural = "analysis grid observations" def __str__(self): return ( f"AnalysisGridObservation({self.cell_id}, " f"{self.temporal_start}, {self.temporal_end})" ) class RemoteSensingSubdivisionResult(models.Model): soil_location = models.ForeignKey( SoilLocation, on_delete=models.CASCADE, related_name="remote_sensing_subdivision_results", ) run = models.OneToOneField( RemoteSensingRun, on_delete=models.CASCADE, related_name="subdivision_result", ) block_subdivision = models.ForeignKey( BlockSubdivision, on_delete=models.SET_NULL, null=True, blank=True, related_name="remote_sensing_subdivision_results", ) block_code = models.CharField( max_length=64, blank=True, default="", db_index=True, ) chunk_size_sqm = models.PositiveIntegerField(default=900) temporal_start = models.DateField(db_index=True) temporal_end = models.DateField(db_index=True) cluster_count = models.PositiveIntegerField(default=0) selected_features = models.JSONField(default=list, blank=True) skipped_cell_codes = models.JSONField(default=list, blank=True) metadata = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["-created_at", "-id"] indexes = [ models.Index( fields=["soil_location", "block_code", "temporal_start", "temporal_end"], name="rs_subdiv_result_lookup_idx", ) ] verbose_name = "remote sensing subdivision result" verbose_name_plural = "remote sensing subdivision results" def __str__(self): return ( f"RemoteSensingSubdivisionResult({self.soil_location_id}, " f"{self.block_code or 'farm'}, clusters={self.cluster_count})" ) class RemoteSensingSubdivisionOption(models.Model): result = models.ForeignKey( RemoteSensingSubdivisionResult, on_delete=models.CASCADE, related_name="options", ) requested_k = models.PositiveIntegerField(db_index=True) effective_cluster_count = models.PositiveIntegerField(default=0) is_active = models.BooleanField(default=False, db_index=True) is_recommended = models.BooleanField(default=False, db_index=True) selection_source = models.CharField( max_length=32, default="system", help_text="منشا انتخاب این گزینه؛ مثل system یا user.", ) metadata = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["result", "requested_k", "id"] constraints = [ models.UniqueConstraint( fields=["result", "requested_k"], name="rs_subdiv_option_unique_result_requested_k", ) ] indexes = [ models.Index( fields=["result", "is_active"], name="rs_subdiv_option_active_idx", ) ] verbose_name = "remote sensing subdivision option" verbose_name_plural = "remote sensing subdivision options" def __str__(self): return ( f"RemoteSensingSubdivisionOption(result={self.result_id}, " f"requested_k={self.requested_k}, active={self.is_active})" ) class RemoteSensingSubdivisionOptionBlock(models.Model): option = models.ForeignKey( RemoteSensingSubdivisionOption, on_delete=models.CASCADE, related_name="cluster_blocks", ) cluster_label = models.PositiveIntegerField(db_index=True) sub_block_code = models.CharField(max_length=64, db_index=True) chunk_size_sqm = models.PositiveIntegerField(default=900) centroid_lat = models.DecimalField(max_digits=9, decimal_places=6, db_index=True) centroid_lon = models.DecimalField(max_digits=9, decimal_places=6, db_index=True) center_cell_code = models.CharField( max_length=64, blank=True, default="", db_index=True, help_text="شناسه سلول مرکزی انتخاب‌شده با بهینه‌سازی 1-center روی اعضای همین کلاستر.", ) center_cell_lat = models.DecimalField( max_digits=9, decimal_places=6, null=True, blank=True, db_index=True, help_text="عرض جغرافیایی سلول مرکزی کلاستر.", ) center_cell_lon = models.DecimalField( max_digits=9, decimal_places=6, null=True, blank=True, db_index=True, help_text="طول جغرافیایی سلول مرکزی کلاستر.", ) geometry = models.JSONField(default=dict, blank=True) cell_count = models.PositiveIntegerField(default=0) cell_codes = models.JSONField(default=list, blank=True) metadata = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["option", "cluster_label", "id"] constraints = [ models.UniqueConstraint( fields=["option", "cluster_label"], name="rs_subdiv_option_block_unique_option_label", ) ] verbose_name = "remote sensing subdivision option block" verbose_name_plural = "remote sensing subdivision option blocks" def __str__(self): return ( f"RemoteSensingSubdivisionOptionBlock(option={self.option_id}, " f"cluster={self.cluster_label})" ) class RemoteSensingSubdivisionOptionAssignment(models.Model): option = models.ForeignKey( RemoteSensingSubdivisionOption, on_delete=models.CASCADE, related_name="assignments", ) cell = models.ForeignKey( AnalysisGridCell, on_delete=models.CASCADE, related_name="subdivision_option_assignments", ) cluster_label = models.PositiveIntegerField(db_index=True) raw_feature_values = models.JSONField(default=dict, blank=True) scaled_feature_values = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["option", "cluster_label", "cell__cell_code"] constraints = [ models.UniqueConstraint( fields=["option", "cell"], name="rs_subdiv_option_assign_unique_option_cell", ) ] indexes = [ models.Index( fields=["option", "cluster_label"], name="rs_subopt_assign_lbl_idx", ) ] verbose_name = "remote sensing subdivision option assignment" verbose_name_plural = "remote sensing subdivision option assignments" def __str__(self): return ( f"RemoteSensingSubdivisionOptionAssignment(option={self.option_id}, " f"cell={self.cell_id}, cluster={self.cluster_label})" ) class RemoteSensingClusterBlock(models.Model): uuid = models.UUIDField(default=uuid.uuid4, editable=False, unique=True, db_index=True) result = models.ForeignKey( RemoteSensingSubdivisionResult, on_delete=models.CASCADE, related_name="cluster_blocks", ) soil_location = models.ForeignKey( SoilLocation, on_delete=models.CASCADE, related_name="remote_sensing_cluster_blocks", ) block_subdivision = models.ForeignKey( BlockSubdivision, on_delete=models.SET_NULL, null=True, blank=True, related_name="remote_sensing_cluster_blocks", ) block_code = models.CharField( max_length=64, blank=True, default="", db_index=True, help_text="شناسه بلوک والد که این زیر‌بلاک KMeans داخل آن ساخته شده است.", ) sub_block_code = models.CharField( max_length=64, db_index=True, help_text="شناسه زیر‌بلاک ساخته‌شده توسط KMeans مثل cluster-0.", ) cluster_label = models.PositiveIntegerField(db_index=True) chunk_size_sqm = models.PositiveIntegerField(default=900) centroid_lat = models.DecimalField( max_digits=9, decimal_places=6, db_index=True, help_text="عرض جغرافیایی مرکز زیر‌بلاک.", ) centroid_lon = models.DecimalField( max_digits=9, decimal_places=6, db_index=True, help_text="طول جغرافیایی مرکز زیر‌بلاک.", ) center_cell_code = models.CharField( max_length=64, blank=True, default="", db_index=True, help_text="شناسه سلول مرکزی انتخاب‌شده با بهینه‌سازی 1-center در همین کلاستر.", ) center_cell_lat = models.DecimalField( max_digits=9, decimal_places=6, null=True, blank=True, db_index=True, help_text="عرض جغرافیایی سلول مرکزی کلاستر.", ) center_cell_lon = models.DecimalField( max_digits=9, decimal_places=6, null=True, blank=True, db_index=True, help_text="طول جغرافیایی سلول مرکزی کلاستر.", ) geometry = models.JSONField( default=dict, blank=True, help_text="هندسه GeoJSON زیر‌بلاک KMeans. فعلا از چندضلعی/چندچندضلعی سلول‌های عضو ساخته می‌شود.", ) cell_count = models.PositiveIntegerField(default=0) cell_codes = models.JSONField(default=list, blank=True) metadata = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["result", "cluster_label", "id"] constraints = [ models.UniqueConstraint( fields=["result", "cluster_label"], name="rs_cluster_block_unique_result_label", ) ] indexes = [ models.Index( fields=["soil_location", "block_code", "cluster_label"], name="rs_cluster_block_lookup_idx", ) ] verbose_name = "remote sensing cluster block" verbose_name_plural = "remote sensing cluster blocks" def __str__(self): return ( f"RemoteSensingClusterBlock({self.uuid}, " f"{self.block_code or 'farm'}:{self.sub_block_code})" ) class RemoteSensingClusterAssignment(models.Model): result = models.ForeignKey( RemoteSensingSubdivisionResult, on_delete=models.CASCADE, related_name="assignments", ) cell = models.ForeignKey( AnalysisGridCell, on_delete=models.CASCADE, related_name="cluster_assignments", ) cluster_label = models.PositiveIntegerField(db_index=True) raw_feature_values = models.JSONField(default=dict, blank=True) scaled_feature_values = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ["cluster_label", "cell__cell_code"] constraints = [ models.UniqueConstraint( fields=["result", "cell"], name="rs_cluster_assign_unique_result_cell", ) ] indexes = [ models.Index( fields=["result", "cluster_label"], name="rs_ca_result_label_idx", ) ] verbose_name = "remote sensing cluster assignment" verbose_name_plural = "remote sensing cluster assignments" def __str__(self): return f"RemoteSensingClusterAssignment({self.result_id}, {self.cell_id}, {self.cluster_label})" class NdviObservation(models.Model): location = models.ForeignKey( SoilLocation, on_delete=models.CASCADE, related_name="ndvi_observations", ) observation_date = models.DateField(db_index=True) mean_ndvi = models.FloatField() ndvi_map = models.JSONField(default=dict, blank=True) vegetation_health_class = models.CharField(max_length=64) satellite_source = models.CharField(max_length=64, default="sentinel-2") cloud_cover = models.FloatField(null=True, blank=True) metadata = models.JSONField(default=dict, blank=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True) class Meta: db_table = "dashboard_data_ndviobservation" ordering = ["-observation_date", "-created_at"] constraints = [ models.UniqueConstraint( fields=["location", "observation_date", "satellite_source"], name="ndvi_unique_location_date_source", ) ] verbose_name = "NDVI Observation" verbose_name_plural = "NDVI Observations" def __str__(self): return f"NDVI {self.location_id} {self.observation_date} {self.satellite_source}"