Skip to content

Sentinel-3 Data Reader

s3lst_ds.data_reading.s3_reader.Sen3Loader dataclass

Data class for Sentinel-3 data

Methods:

Name Description
__init__
__post_init__

Load Sentinel-3 data and extract LST's starting sensing time
calc_fvc

Calculate the Fractional Vegetation Cover (FVC) from Sentinel-3 OLCI SYN
calc_ndvi

Calculate NDVI from Sentinel-3 SYN surface directional reflectances associated
calc_ndwi

Calculate the Normalized Difference Water Index from Sentinel-3 OLCI SYN surface
get_var_by_name

Fetches and returns variables based on their name.

Attributes:

Name Type Description
lst Dataset
path Path
syn Dataset
timestamp Timestamp
Source code in src/s3lst_ds/data_reading/s3_reader.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@dataclass
class Sen3Loader:
    """Data class for Sentinel-3 data"""

    path: Path
    lst: xr.Dataset = field(init=False)
    syn: xr.Dataset = field(init=False)
    timestamp: pd.Timestamp = field(init=False)

    def __post_init__(self):
        """Load Sentinel-3 data and extract LST's starting sensing time"""
        try:
            self.lst = self._load_file(next(self.path.glob("*LST*.SEN3.nc")))
            self.syn = self._load_file(next(self.path.glob("*SYN*.SEN3.nc")))
            self.timestamp = self._get_timestamp()

        # NOTE: There is no error message in the case of an StopIteration exception
        # (which occurs when in the next() function there is no element to iterate over
        # (which means that the file is missing)). This is then herein handled in a
        # way that is different from general exception's.
        except StopIteration:
            raise FailedProdReadingError("Sentinel-3 LST or SYN files not found.")

        except Exception as e:  # noqa: BLE001
            raise FailedProdReadingError(
                "Error reading Sentinel-3 LST or SYN files." + f"\nError message: {e}"
            )

    def _load_file(self, field_path) -> xr.Dataset:
        """Load Sentinel-3 LST or SYN netCDF files from directory"""

        data = xr.open_dataset(
            field_path,  # type: ignore
            # Mask out NODATA values and scale
            mask_and_scale=True,
            # Properly decode coordinates and CRS
            decode_coords="all",
        )

        if "lon" in data.coords:
            data = data.rename({"lon": "x"})

        if "lat" in data.coords:
            data = data.rename({"lat": "y"})

        return data

    def _get_timestamp(self) -> pd.Timestamp:
        """
        Extract the start sensing timestamp from the Sentinel-3 LST product name.

        Returns
        -------

        timestamp : pd.Timestamp
            Start sensing timestamp associated with Sentinel-3 LST product.

        """
        timestamp = pd.Timestamp(next(self.path.glob("*LST*.SEN3.nc")).name[16:31])

        return timestamp

    def calc_ndvi(self) -> xr.DataArray:
        """
        Calculate NDVI from Sentinel-3 SYN surface directional reflectances associated
        with the OLCI channels [08, 17].
        """
        # Compute NDVI
        # NOTE: the result is an unamed DataArray whose data corresponds to the
        # result of the operation on the data of the raw DataArrays, having the same
        # attributes "dims" and "coords". But the remainder of the attributes would
        # be undefined. For instance, "attrs", "rio.nodata" and "rio.encoded_nodata"
        # would correspond to the respective default values ({}, None and None,
        # respectively.)
        ndvi = (self.syn.SDR_Oa17 - self.syn.SDR_Oa08) / (
            self.syn.SDR_Oa17 + self.syn.SDR_Oa08
        )

        ndvi = ndvi.rename("NDVI")
        ndvi.attrs = self.syn.SDR_Oa08.attrs
        ndvi.attrs["long_name"] = "Normalized Difference Vegetation Index"
        ndvi.attrs["standard_name"] = "normalized_difference_vegetation_index"
        ndvi.attrs["title"] = (
            "SYN L2, NDVI from surface directional reflectance associated with OLCI"
            + " channels [08, 17]"
        )
        ndvi.attrs["units"] = None
        ndvi.attrs["valid_max"] = np.nanmax(ndvi.data)
        ndvi.attrs["valid_min"] = np.nanmin(ndvi.data)
        for key in SYN_ATTRS_TO_REMOVE:
            ndvi.attrs.pop(key, None)

        # Set the NODATA value
        # NOTE: the data of ndvi is already encoded (that is, the NODATA instances
        # correspond to nan) since the respective computation bands (a08 and a17) were
        # also encoded. One may set the encoding NODATA value (the one used in the
        # exported files) through method rio.write_nodata() with encoded=True.
        ndvi.rio.write_nodata(-32768, encoded=True, inplace=True)

        return ndvi

    def calc_ndwi(self) -> xr.DataArray:
        """
        Calculate the Normalized Difference Water Index from Sentinel-3 OLCI SYN surface
        directional reflectance product [channels: 06, 17].
        """

        ndwi = (self.syn.SDR_Oa06 - self.syn.SDR_Oa17) / (
            self.syn.SDR_Oa06 + self.syn.SDR_Oa17
        )

        ndwi = ndwi.rename("NDWI")
        ndwi.attrs = self.syn.SDR_Oa06.attrs
        ndwi.attrs["long_name"] = "Normalized Difference Water Index"
        ndwi.attrs["standard_name"] = "normalized_difference_water_index"
        ndwi.attrs["title"] = (
            "SYN L2, NDWI from surface directional reflectance associated with OLCI"
            + " channels [06, 17]"
        )
        ndwi.attrs["units"] = None
        ndwi.attrs["valid_max"] = np.nanmax(ndwi.data)
        ndwi.attrs["valid_min"] = np.nanmin(ndwi.data)
        for key in SYN_ATTRS_TO_REMOVE:
            ndwi.attrs.pop(key, None)

        ndwi.rio.write_nodata(-32768, encoded=True, inplace=True)

        return ndwi

    def calc_fvc(self) -> xr.DataArray:
        """
        Calculate the Fractional Vegetation Cover (FVC) from Sentinel-3 OLCI SYN
        product. The FVC is calculated using the NDVI and the NDVI thresholds for bare
        soil and full vegetation cover.
        """
        # Compute FVC from NDVI
        ndvi = self.calc_ndvi()
        ndvi_min = np.nanmin(ndvi.values)
        ndvi_max = np.nanmax(ndvi.values)

        # NOTE: see equation 7 of wang2022 (doi: 10.3390/rs14225752)
        fvc = 1 - ((ndvi_max - ndvi) / (ndvi_max - ndvi_min)) ** 0.625

        fvc = fvc.rename("FVC")
        fvc.attrs = ndvi.attrs
        fvc.attrs["long_name"] = "Fractional Vegetation Cover"
        fvc.attrs["standard_name"] = "fractional_vegetation_cover"
        fvc.attrs["title"] = "SYN L2, FVC from NDVI"

        fvc.attrs["units"] = None
        fvc.attrs["valid_max"] = np.nanmax(fvc.data)
        fvc.attrs["valid_min"] = np.nanmin(fvc.data)
        for key in SYN_ATTRS_TO_REMOVE:
            fvc.attrs.pop(key, None)

        fvc.rio.write_nodata(-32768, encoded=True, inplace=True)

        return fvc

    def get_var_by_name(self, var: str) -> xr.DataArray:
        """
        Fetches and returns variables based on their name.
        """
        match var.lower():
            case "lst":
                return self.lst.LST.to_dataset()
            case "ndvi":
                return self.calc_ndvi()
            case "ndwi":
                return self.calc_ndwi()
            case "fvc":
                return self.calc_fvc()
            case _:
                raise ValueError(
                    f"Unknown variable name: {var}. "
                    "Available name: 'LST','NDVI', 'NDWI', 'FVC'."
                )

lst class-attribute instance-attribute 

lst: Dataset = field(init=False)

path instance-attribute 

path: Path

syn class-attribute instance-attribute 

syn: Dataset = field(init=False)

timestamp class-attribute instance-attribute 

timestamp: Timestamp = field(init=False)

__init__ 

__init__(path: Path) -> None

__post_init__ 

__post_init__()

Load Sentinel-3 data and extract LST's starting sensing time

Source code in src/s3lst_ds/data_reading/s3_reader.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __post_init__(self):
    """Load Sentinel-3 data and extract LST's starting sensing time"""
    try:
        self.lst = self._load_file(next(self.path.glob("*LST*.SEN3.nc")))
        self.syn = self._load_file(next(self.path.glob("*SYN*.SEN3.nc")))
        self.timestamp = self._get_timestamp()

    # NOTE: There is no error message in the case of an StopIteration exception
    # (which occurs when in the next() function there is no element to iterate over
    # (which means that the file is missing)). This is then herein handled in a
    # way that is different from general exception's.
    except StopIteration:
        raise FailedProdReadingError("Sentinel-3 LST or SYN files not found.")

    except Exception as e:  # noqa: BLE001
        raise FailedProdReadingError(
            "Error reading Sentinel-3 LST or SYN files." + f"\nError message: {e}"
        )

calc_fvc 

calc_fvc() -> DataArray

Calculate the Fractional Vegetation Cover (FVC) from Sentinel-3 OLCI SYN product. The FVC is calculated using the NDVI and the NDVI thresholds for bare soil and full vegetation cover.

Source code in src/s3lst_ds/data_reading/s3_reader.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def calc_fvc(self) -> xr.DataArray:
    """
    Calculate the Fractional Vegetation Cover (FVC) from Sentinel-3 OLCI SYN
    product. The FVC is calculated using the NDVI and the NDVI thresholds for bare
    soil and full vegetation cover.
    """
    # Compute FVC from NDVI
    ndvi = self.calc_ndvi()
    ndvi_min = np.nanmin(ndvi.values)
    ndvi_max = np.nanmax(ndvi.values)

    # NOTE: see equation 7 of wang2022 (doi: 10.3390/rs14225752)
    fvc = 1 - ((ndvi_max - ndvi) / (ndvi_max - ndvi_min)) ** 0.625

    fvc = fvc.rename("FVC")
    fvc.attrs = ndvi.attrs
    fvc.attrs["long_name"] = "Fractional Vegetation Cover"
    fvc.attrs["standard_name"] = "fractional_vegetation_cover"
    fvc.attrs["title"] = "SYN L2, FVC from NDVI"

    fvc.attrs["units"] = None
    fvc.attrs["valid_max"] = np.nanmax(fvc.data)
    fvc.attrs["valid_min"] = np.nanmin(fvc.data)
    for key in SYN_ATTRS_TO_REMOVE:
        fvc.attrs.pop(key, None)

    fvc.rio.write_nodata(-32768, encoded=True, inplace=True)

    return fvc

calc_ndvi 

calc_ndvi() -> DataArray

Calculate NDVI from Sentinel-3 SYN surface directional reflectances associated with the OLCI channels [08, 17].

Source code in src/s3lst_ds/data_reading/s3_reader.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def calc_ndvi(self) -> xr.DataArray:
    """
    Calculate NDVI from Sentinel-3 SYN surface directional reflectances associated
    with the OLCI channels [08, 17].
    """
    # Compute NDVI
    # NOTE: the result is an unamed DataArray whose data corresponds to the
    # result of the operation on the data of the raw DataArrays, having the same
    # attributes "dims" and "coords". But the remainder of the attributes would
    # be undefined. For instance, "attrs", "rio.nodata" and "rio.encoded_nodata"
    # would correspond to the respective default values ({}, None and None,
    # respectively.)
    ndvi = (self.syn.SDR_Oa17 - self.syn.SDR_Oa08) / (
        self.syn.SDR_Oa17 + self.syn.SDR_Oa08
    )

    ndvi = ndvi.rename("NDVI")
    ndvi.attrs = self.syn.SDR_Oa08.attrs
    ndvi.attrs["long_name"] = "Normalized Difference Vegetation Index"
    ndvi.attrs["standard_name"] = "normalized_difference_vegetation_index"
    ndvi.attrs["title"] = (
        "SYN L2, NDVI from surface directional reflectance associated with OLCI"
        + " channels [08, 17]"
    )
    ndvi.attrs["units"] = None
    ndvi.attrs["valid_max"] = np.nanmax(ndvi.data)
    ndvi.attrs["valid_min"] = np.nanmin(ndvi.data)
    for key in SYN_ATTRS_TO_REMOVE:
        ndvi.attrs.pop(key, None)

    # Set the NODATA value
    # NOTE: the data of ndvi is already encoded (that is, the NODATA instances
    # correspond to nan) since the respective computation bands (a08 and a17) were
    # also encoded. One may set the encoding NODATA value (the one used in the
    # exported files) through method rio.write_nodata() with encoded=True.
    ndvi.rio.write_nodata(-32768, encoded=True, inplace=True)

    return ndvi

calc_ndwi 

calc_ndwi() -> DataArray

Calculate the Normalized Difference Water Index from Sentinel-3 OLCI SYN surface directional reflectance product [channels: 06, 17].

Source code in src/s3lst_ds/data_reading/s3_reader.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def calc_ndwi(self) -> xr.DataArray:
    """
    Calculate the Normalized Difference Water Index from Sentinel-3 OLCI SYN surface
    directional reflectance product [channels: 06, 17].
    """

    ndwi = (self.syn.SDR_Oa06 - self.syn.SDR_Oa17) / (
        self.syn.SDR_Oa06 + self.syn.SDR_Oa17
    )

    ndwi = ndwi.rename("NDWI")
    ndwi.attrs = self.syn.SDR_Oa06.attrs
    ndwi.attrs["long_name"] = "Normalized Difference Water Index"
    ndwi.attrs["standard_name"] = "normalized_difference_water_index"
    ndwi.attrs["title"] = (
        "SYN L2, NDWI from surface directional reflectance associated with OLCI"
        + " channels [06, 17]"
    )
    ndwi.attrs["units"] = None
    ndwi.attrs["valid_max"] = np.nanmax(ndwi.data)
    ndwi.attrs["valid_min"] = np.nanmin(ndwi.data)
    for key in SYN_ATTRS_TO_REMOVE:
        ndwi.attrs.pop(key, None)

    ndwi.rio.write_nodata(-32768, encoded=True, inplace=True)

    return ndwi

get_var_by_name 

get_var_by_name(var: str) -> DataArray

Fetches and returns variables based on their name.

Source code in src/s3lst_ds/data_reading/s3_reader.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def get_var_by_name(self, var: str) -> xr.DataArray:
    """
    Fetches and returns variables based on their name.
    """
    match var.lower():
        case "lst":
            return self.lst.LST.to_dataset()
        case "ndvi":
            return self.calc_ndvi()
        case "ndwi":
            return self.calc_ndwi()
        case "fvc":
            return self.calc_fvc()
        case _:
            raise ValueError(
                f"Unknown variable name: {var}. "
                "Available name: 'LST','NDVI', 'NDWI', 'FVC'."
            )