Spaces:
Paused
Paused
| """Resolve spectral roles from evidence, never from an unqualified band number.""" | |
| from dataclasses import dataclass, field, asdict | |
| import re | |
| def normalize(value): | |
| return re.sub(r"[^a-z0-9]", "", str(value).lower()) | |
| ALIASES = { | |
| "blue": {"blue"}, "green": {"green"}, "red": {"red"}, | |
| "nir": {"nir", "nearinfrared", "nir08"}, | |
| "narrow_nir": {"narrownir", "nirnarrow", "nir09", "b8a"}, | |
| "swir": {"swir", "swir1", "swir16", "shortwaveinfrared1"}, | |
| "swir2": {"swir2", "swir22", "shortwaveinfrared2"}, | |
| "vv": {"vv"}, "vh": {"vh"}, "hh": {"hh"}, "hv": {"hv"}, | |
| } | |
| S2 = {"b2":"blue", "b3":"green", "b4":"red", "b8":"nir", "b8a":"narrow_nir", "b11":"swir", "b12":"swir2"} | |
| LANDSAT_OLI = {"b2":"blue", "b3":"green", "b4":"red", "b5":"nir", "b6":"swir", "b7":"swir2"} | |
| LANDSAT_TM = {"b1":"blue", "b2":"green", "b3":"red", "b4":"nir", "b5":"swir", "b7":"swir2"} | |
| class BandEntry: | |
| band_name: str | |
| raster_index: int | |
| wavelength: float | None = None | |
| confidence: float = 1.0 | |
| source: str = "metadata" | |
| def to_dict(self): | |
| return {**asdict(self), "band_index": self.raster_index} | |
| class BandMap: | |
| indices: dict = field(default_factory=dict) | |
| confidence: dict = field(default_factory=dict) | |
| evidence: dict = field(default_factory=dict) | |
| wavelengths_nm: dict = field(default_factory=dict) | |
| warnings: list = field(default_factory=list) | |
| bands: dict = field(default_factory=dict) | |
| def has_band(self, name: str) -> bool: | |
| return name.lower() in self.indices | |
| def get_index(self, name: str) -> int | None: | |
| return self.indices.get(name.lower()) | |
| def to_dict(self): | |
| d = asdict(self) | |
| d["bands"] = {k: v.to_dict() if hasattr(v, "to_dict") else v for k, v in self.bands.items()} | |
| return d | |
| def detect_band_map(src): | |
| result = BandMap() | |
| tags = {normalize(k): v for k, v in src.tags().items()} | |
| sensor = normalize(" ".join(str(tags.get(k, "")) for k in ("sensor", "satellite", "platform"))) | |
| declared_order = [item.strip() for item in str(tags.get("bandorder", "")).split(",")] | |
| if len(declared_order) != src.count: | |
| declared_order = [] | |
| description_tokens = {normalize(value) for value in (*src.descriptions, *declared_order) if value} | |
| numbered_descriptions = {f"b{int(match.group(1))}" for token in description_tokens | |
| if (match := re.fullmatch(r"(?:band|b)0*(\d+)", token))} | |
| if "sentinel2" not in sensor and ( | |
| "b8a" in description_tokens | |
| or ({"b10", "b11", "b12"} <= description_tokens and len(description_tokens) >= 8) | |
| or {"b2", "b3", "b4", "b8", "b11", "b12"} <= numbered_descriptions | |
| ): | |
| sensor = f"{sensor} sentinel2_inferred_from_band_names" | |
| numbered = S2 if "sentinel2" in sensor else LANDSAT_OLI if any(x in sensor for x in ("landsat8", "landsat9")) else LANDSAT_TM if any(x in sensor for x in ("landsat4", "landsat5", "landsat7")) else {} | |
| candidates = {} | |
| exact_s2_candidates = {} | |
| conflicts = set() | |
| for i in range(1, src.count + 1): | |
| band_tags = {normalize(k): v for k, v in src.tags(i).items()} | |
| names = [src.descriptions[i - 1] or ""] + ([declared_order[i - 1]] if declared_order else []) + [band_tags[k] for k in ("name", "commonname", "bandname", "polarization", "polarisation") if k in band_tags] | |
| # Color interpretation is authoritative for ordinary RGB files. | |
| color = src.colorinterp[i - 1].name | |
| if color in ("red", "green", "blue"): | |
| names.append(color) | |
| roles = set() | |
| reasons = [] | |
| for name in names: | |
| key = normalize(name) | |
| s2_match = re.fullmatch(r"(?:band|b)0*(1[0-2]|[1-9]|8a)", key) | |
| if "sentinel2" in sensor and s2_match: | |
| token = s2_match.group(1) | |
| exact_name = "b8a" if token == "8a" else f"b{int(token):02d}" | |
| exact_s2_candidates.setdefault(exact_name, []).append((i, f"Sentinel-2 metadata: {name}")) | |
| for role, aliases in ALIASES.items(): | |
| if key in aliases: | |
| roles.add(role) | |
| if role == "swir": | |
| roles.add("swir1") | |
| reasons.append(f"band {i}: {name}") | |
| match = re.fullmatch(r"(?:band|b)0*(\d+)", key) | |
| if match: | |
| role = numbered.get("b" + match[1]) | |
| if role: | |
| roles.add(role) | |
| if role == "swir": | |
| roles.add("swir1") | |
| reasons.append(f"{sensor}: {name}") | |
| wave = band_tags.get("wavelength", band_tags.get("centralwavelength")) | |
| unit = normalize(band_tags.get("wavelengthunits", band_tags.get("wavelengthunit", ""))) | |
| if wave is not None and unit in {"nm", "nanometer", "nanometers", "um", "micrometer", "micrometers"}: | |
| try: | |
| nm = float(wave) * (1000 if unit in {"um", "micrometer", "micrometers"} else 1) | |
| result.wavelengths_nm[str(i)] = nm | |
| for role, lo, hi in [("blue", 450, 510), ("green", 520, 600), ("red", 630, 690), ("narrow_nir", 855, 890), ("nir", 780, 854), ("swir", 1550, 1750), ("swir2", 2080, 2350)]: | |
| if lo <= nm <= hi: | |
| roles.add(role) | |
| if role == "swir": | |
| roles.add("swir1") | |
| reasons.append(f"wavelength {nm:g} nm") | |
| except (ValueError, TypeError): | |
| result.warnings.append(f"Invalid wavelength for band {i}.") | |
| if len(roles) > 1 and not (roles == {"swir", "swir1"}): | |
| conflicts.update(roles) | |
| result.warnings.append(f"Conflicting role evidence for band {i}: {sorted(roles)}.") | |
| for role in roles: | |
| candidates.setdefault(role, []).append((i, reasons)) | |
| for role, choices in candidates.items(): | |
| unique_indices = {choice[0] for choice in choices} | |
| if role in conflicts or len(unique_indices) != 1: | |
| result.warnings.append(f"Ambiguous {role} mapping; dependent operations are blocked.") | |
| continue | |
| idx = next(iter(unique_indices)) | |
| reasons_str = "; ".join(dict.fromkeys(reason for _, reasons in choices for reason in reasons)) | |
| result.indices[role] = idx | |
| result.confidence[role] = 1.0 # evidence completeness, not a calibrated probability | |
| result.evidence[role] = choices[0][1] | |
| wavelength_val = result.wavelengths_nm.get(str(idx)) | |
| result.bands[role] = BandEntry( | |
| band_name=role.upper(), | |
| raster_index=idx, | |
| wavelength=wavelength_val, | |
| confidence=1.0, | |
| source=reasons_str or "metadata", | |
| ) | |
| for role, choices in exact_s2_candidates.items(): | |
| unique_indices = {choice[0] for choice in choices} | |
| if len(unique_indices) != 1: | |
| result.warnings.append(f"Ambiguous exact Sentinel-2 {role.upper()} mapping; model inference is blocked.") | |
| continue | |
| index = next(iter(unique_indices)) | |
| result.indices[role] = index | |
| result.confidence[role] = 1.0 | |
| result.evidence[role] = [choice[1] for choice in choices] | |
| result.bands[role] = BandEntry( | |
| band_name=role.upper(), raster_index=index, confidence=1.0, | |
| source="; ".join(choice[1] for choice in choices), | |
| ) | |
| # Ensure swir and swir1 cross-alias | |
| if "swir" in result.indices and "swir1" not in result.indices: | |
| result.indices["swir1"] = result.indices["swir"] | |
| result.confidence["swir1"] = result.confidence["swir"] | |
| result.evidence["swir1"] = result.evidence["swir"] | |
| elif "swir1" in result.indices and "swir" not in result.indices: | |
| result.indices["swir"] = result.indices["swir1"] | |
| result.confidence["swir"] = result.confidence["swir1"] | |
| result.evidence["swir"] = result.evidence["swir1"] | |
| return result | |