| | |
| | | MIN_MAX_DIMENSION = 512 |
| | | MAX_MAX_DIMENSION = 4096 |
| | | PROCESSING_MODES = {"auto", "image", "geotiff"} |
| | | GENERIC_DIFF_THRESHOLD = 0.45 |
| | | |
| | | |
| | | def _to_uint8(data: np.ndarray) -> np.ndarray: |
| | |
| | | |
| | | |
| | | def _write_overlay(path: Path, before: np.ndarray, after: np.ndarray, mask: np.ndarray) -> None: |
| | | _write_evidence_overlay(path, before, after, mask, None) |
| | | |
| | | |
| | | def _write_evidence_overlay(path: Path, before: np.ndarray, after: np.ndarray, model_mask: np.ndarray, generic_mask: np.ndarray | None) -> None: |
| | | left = before.copy() |
| | | right = after.copy() |
| | | if generic_mask is not None: |
| | | cyan = np.zeros_like(right) |
| | | cyan[..., 1] = 220 |
| | | cyan[..., 2] = 255 |
| | | right[generic_mask] = (right[generic_mask].astype(np.float32) * 0.45 + cyan[generic_mask].astype(np.float32) * 0.55).astype(np.uint8) |
| | | red = np.zeros_like(right) |
| | | red[..., 0] = 255 |
| | | right[mask] = (right[mask].astype(np.float32) * 0.45 + red[mask].astype(np.float32) * 0.55).astype(np.uint8) |
| | | right[model_mask] = (right[model_mask].astype(np.float32) * 0.45 + red[model_mask].astype(np.float32) * 0.55).astype(np.uint8) |
| | | separator = np.full((before.shape[0], 8, 3), 235, dtype=np.uint8) |
| | | Image.fromarray(np.concatenate([left, separator, right], axis=1)).save(path, quality=92) |
| | | |
| | | |
| | | def _compute_generic_difference_mask(before: np.ndarray, after: np.ndarray, valid: np.ndarray, *, threshold: float = GENERIC_DIFF_THRESHOLD) -> tuple[np.ndarray, dict[str, Any]]: |
| | | """Create a conservative visual-difference candidate layer. |
| | | |
| | | This is deliberately separate from ChangeStar's semantic mask: it highlights |
| | | strong local RGB differences such as moved vehicles or people, while |
| | | removing tiny noise, very large illumination regions, and invalid borders. |
| | | """ |
| | | if before.shape != after.shape or before.shape[:2] != valid.shape: |
| | | raise ValueError("before, after, and valid must share the same spatial shape") |
| | | difference = np.mean(np.abs(before.astype(np.float32) - after.astype(np.float32)), axis=2) / 255.0 |
| | | difference = cv2.GaussianBlur(difference, (3, 3), 0) |
| | | candidate = ((difference >= threshold) & valid).astype(np.uint8) |
| | | candidate = cv2.morphologyEx(candidate, cv2.MORPH_OPEN, np.ones((3, 3), dtype=np.uint8)) |
| | | candidate = cv2.morphologyEx(candidate, cv2.MORPH_CLOSE, np.ones((5, 5), dtype=np.uint8)) |
| | | count, labels, stats, _ = cv2.connectedComponentsWithStats(candidate, connectivity=8) |
| | | min_area = max(32, int(candidate.size * 0.00003)) |
| | | max_area = max(min_area, int(candidate.size * 0.08)) |
| | | cleaned = np.zeros_like(candidate) |
| | | kept = 0 |
| | | for component in range(1, count): |
| | | x, y, width, height, area = (int(value) for value in stats[component]) |
| | | touches_edge = x <= 0 or y <= 0 or x + width >= candidate.shape[1] or y + height >= candidate.shape[0] |
| | | if min_area <= area <= max_area and not touches_edge: |
| | | cleaned[labels == component] = 1 |
| | | kept += 1 |
| | | details = { |
| | | "threshold": threshold, |
| | | "minimum_component_pixels": min_area, |
| | | "maximum_component_pixels": max_area, |
| | | "components_before_filter": max(0, count - 1), |
| | | "components_after_filter": kept, |
| | | "candidate_pixels": int(cleaned.sum()), |
| | | "candidate_pixel_ratio": round(float(cleaned.mean()), 6), |
| | | } |
| | | return cleaned > 0, details |
| | | |
| | | |
| | | def _feature_summary(mask: np.ndarray, probability: np.ndarray, transform: Affine, georeferenced: bool) -> list[dict[str, Any]]: |
| | |
| | | output_crs = before_info["crs"] if actual_mode == "geotiff" else None |
| | | _write_rgb_geotiff(work_dir / "before.tif", before_small, transform, output_crs) |
| | | _write_rgb_geotiff(work_dir / "after_registered.tif", after_small, transform, output_crs) |
| | | Image.fromarray(before_small).save(output_dir / "before_processed_preview.jpg", quality=92) |
| | | Image.fromarray(after_small).save(output_dir / "after_registered_preview.jpg", quality=92) |
| | | |
| | | from geoai import ChangeStarDetection |
| | | |
| | |
| | | for component in range(1, count): |
| | | if int(stats[component, cv2.CC_STAT_AREA]) < min_area: |
| | | cleaned[labels == component] = 0 |
| | | generic_mask, generic_details = _compute_generic_difference_mask(before_small, after_small, valid_small) |
| | | _write_raster(output_dir / "change_probability.tif", probability, transform, "float32", output_crs) |
| | | _write_raster(output_dir / "change_mask_raw.tif", raw_mask, transform, "uint8", output_crs) |
| | | _write_raster(output_dir / "change_mask.tif", cleaned, transform, "uint8", output_crs) |
| | | _write_overlay(output_dir / "change_overlay.jpg", before_small, after_small, cleaned > 0) |
| | | _write_raster(output_dir / "generic_difference_mask.tif", generic_mask.astype(np.uint8) * 255, transform, "uint8", output_crs) |
| | | _write_evidence_overlay(output_dir / "change_overlay.jpg", before_small, after_small, cleaned > 0, generic_mask) |
| | | _write_overlay(output_dir / "change_model_overlay.jpg", before_small, after_small, cleaned > 0) |
| | | vector_path = output_dir / "changes.geojson" |
| | | features, vector_count = vectorize_cleaned_mask(output_dir / "change_mask.tif", output_dir / "change_probability.tif", vector_path) |
| | | from rectangularize_vectors import rectangularize_vector |
| | |
| | | "raw_changed_pixels": int((raw_mask > 0).sum()), |
| | | "changed_pixels": int((cleaned > 0).sum()), |
| | | "changed_pixel_ratio": round(float((cleaned > 0).mean()), 6), |
| | | "generic_difference": generic_details, |
| | | "vector_feature_count": vector_count, |
| | | "rectangle_feature_count": rectangle_count, |
| | | "georeferenced": actual_mode == "geotiff", |
| | |
| | | "probability_raster": "change_probability.tif", |
| | | "raw_mask_raster": "change_mask_raw.tif", |
| | | "mask_raster": "change_mask.tif", |
| | | "generic_difference_mask": "generic_difference_mask.tif", |
| | | "overlay": "change_overlay.jpg", |
| | | "model_overlay": "change_model_overlay.jpg", |
| | | "before_processed_preview": "before_processed_preview.jpg", |
| | | "after_registered_preview": "after_registered_preview.jpg", |
| | | "vector": "changes.geojson", |
| | | "rectangle_vector": "changes_rectangles.geojson", |
| | | "rectangle_vector_wgs84": "changes_rectangles_wgs84.geojson" if actual_mode == "geotiff" else None, |