class Sentinel3SlstrRBT(Copernicus):
"""Sentinel-3 SLSTR Level-1 nadir reflectance and brightness temperature."""
PRODUCT_TYPE = "SL_1_RBT___"
DEFAULT_BANDS = SLSTR_REFLECTANCE_BANDS + SLSTR_BT_BANDS
def __init__(
self,
band_names: list[str] | None = None,
reflectance_grid_resolution: float = 0.0045,
bt_grid_resolution: float = 0.009,
nodata_value: float = -9999.0,
swath_padding: float = 0.1,
context: DataSourceContext = DataSourceContext(),
**kwargs: Any,
) -> None:
"""Create an SLSTR RBT source.
Args:
band_names: bands to ingest when no layer configuration is available.
reflectance_grid_resolution: intermediate WGS84 resolution for S1-S6.
bt_grid_resolution: intermediate WGS84 resolution for S7-S9.
nodata_value: value assigned outside the valid swath.
swath_padding: padding in degrees around requested windows before gridding.
context: data source context.
kwargs: additional options for :class:`Copernicus`.
"""
if reflectance_grid_resolution <= 0 or bt_grid_resolution <= 0:
raise ValueError("grid resolutions must be positive")
if swath_padding < 0:
raise ValueError("swath_padding cannot be negative")
self.band_names = _requested_bands(context, band_names, self.DEFAULT_BANDS)
self.reflectance_bands = [
band for band in SLSTR_REFLECTANCE_BANDS if band in self.band_names
]
self.bt_bands = [band for band in SLSTR_BT_BANDS if band in self.band_names]
self.reflectance_grid_resolution = reflectance_grid_resolution
self.bt_grid_resolution = bt_grid_resolution
self.nodata_value = nodata_value
self.swath_padding = swath_padding
band_groups = {}
if self.reflectance_bands:
band_groups["SLSTR_RBT_REFLECTANCE"] = self.reflectance_bands
if self.bt_bands:
band_groups["SLSTR_RBT_BT"] = self.bt_bands
super().__init__(
glob_to_bands=band_groups,
query_filter=_sentinel3_query_filter("SLSTR", self.PRODUCT_TYPE),
context=context,
**kwargs,
)
def _process_product_zip(
self,
tile_store: TileStoreWithLayer,
item: CopernicusItem,
local_zip_fname: str,
geometries: list[STGeometry] | None = None,
) -> None:
"""Convert and ingest requested SLSTR reflectance and thermal bands."""
needs_reflectance = bool(
self.reflectance_bands
) and not tile_store.is_raster_ready(item, self.reflectance_bands)
needs_bt = bool(self.bt_bands) and not tile_store.is_raster_ready(
item, self.bt_bands
)
if not needs_reflectance and not needs_bt:
return
required: set[str] = set()
if needs_reflectance:
required.update(
{
"indices_an.nc",
"geometry_tn.nc",
"geodetic_an.nc",
"cartesian_tx.nc",
"cartesian_an.nc",
}
)
for band_name in self.reflectance_bands:
band = band_name.split("_")[0]
required.update({f"{band}_radiance_an.nc", f"{band}_quality_an.nc"})
if needs_bt:
required.add("geodetic_in.nc")
required.update(f"{band}_in.nc" for band in self.bt_bands)
with tempfile.TemporaryDirectory() as tmp_dir, ZipFile(local_zip_fname) as zipf:
paths = _extract_zip_members(zipf, required, tmp_dir)
if needs_reflectance:
detector = _read_netcdf_variable(
paths["indices_an.nc"], "detector_an"
).astype(np.float32)
valid_detector = np.isfinite(detector) & (detector >= 0)
safe_detector = np.clip(
np.nan_to_num(detector, nan=0.0).astype(np.int64), 0, None
)
image_x = _read_netcdf_variable(paths["cartesian_an.nc"], "x_an")
image_y = _read_netcdf_variable(paths["cartesian_an.nc"], "y_an")
solar_zenith = _interpolate_tie_points_xy(
_read_netcdf_variable(paths["geometry_tn.nc"], "solar_zenith_tn"),
_read_netcdf_variable(paths["cartesian_tx.nc"], "x_tx"),
_read_netcdf_variable(paths["cartesian_tx.nc"], "y_tx"),
image_x,
image_y,
)
cos_solar_zenith = np.clip(
np.cos(np.deg2rad(solar_zenith)), 0.01, None
).astype(np.float32)
reflectance_arrays = []
for band_name in self.reflectance_bands:
band = band_name.split("_")[0]
radiance_name = f"{band}_radiance_an"
irradiance_name = f"{band}_solar_irradiance_an"
radiance = _read_netcdf_variable(
paths[f"{radiance_name}.nc"], radiance_name
).astype(np.float32)
detector_irradiance = _read_netcdf_variable(
paths[f"{band}_quality_an.nc"], irradiance_name
).astype(np.float32)
band_detector = np.clip(
safe_detector, 0, detector_irradiance.size - 1
)
reflectance = _radiance_to_reflectance(
radiance,
detector_irradiance.reshape(-1)[band_detector],
cos_solar_zenith,
)
reflectance_arrays.append(
np.where(valid_detector, reflectance, np.nan)
)
reflectance_latitude = _read_netcdf_variable(
paths["geodetic_an.nc"], "latitude_an"
).astype(np.float32)
reflectance_longitude = _read_netcdf_variable(
paths["geodetic_an.nc"], "longitude_an"
).astype(np.float32)
if needs_bt:
bt_arrays = []
for band in self.bt_bands:
variable_name = f"{band}_in"
bt_arrays.append(
_read_netcdf_variable(
paths[f"{variable_name}.nc"], variable_name
).astype(np.float32)
)
bt_latitude = _read_netcdf_variable(
paths["geodetic_in.nc"], "latitude_in"
).astype(np.float32)
bt_longitude = _read_netcdf_variable(
paths["geodetic_in.nc"], "longitude_in"
).astype(np.float32)
if needs_reflectance:
data, longitude, latitude = _crop_swath(
np.stack(reflectance_arrays),
reflectance_longitude,
reflectance_latitude,
geometries,
self.swath_padding,
)
_write_swath(
tile_store,
item,
self.reflectance_bands,
data,
longitude,
latitude,
self.reflectance_grid_resolution,
self.nodata_value,
)
if needs_bt:
data, longitude, latitude = _crop_swath(
np.stack(bt_arrays),
bt_longitude,
bt_latitude,
geometries,
self.swath_padding,
)
_write_swath(
tile_store,
item,
self.bt_bands,
data,
longitude,
latitude,
self.bt_grid_resolution,
self.nodata_value,
)