Skip to content

Geoprocessor Modis

MODISGeoProcessing dataclass

Source code in rs_tools/_src/geoprocessing/modis/geoprocessor_modis.py 
 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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
@dataclass
class MODISGeoProcessing:
    satellite: str
    read_path: str
    save_path: str
    """
    A class for geoprocessing MODIS data.

    Attributes:
        satellite (str): The satellite to geoprocess data for.
        read_path (str): The path to read the files from.
        save_path (str): The path to save the processed files to.

    Methods:
        modis_files(self) -> List[str]: Returns a list of all MODIS files in the read path.
        preprocess_fn(self, ds: xr.Dataset) -> Tuple[xr.Dataset, xr.Dataset]: Preprocesses the input dataset by applying corrections.
        preprocess_fn_radiances(self, ds: xr.Dataset) -> xr.Dataset: Preprocesses the input dataset for radiances.
        preprocess_fn_cloudmask(self, ds: xr.Dataset) -> xr.Dataset: Preprocesses the input dataset for cloud mask.
        preprocess_files(self): Preprocesses the files in the read path and saves the processed files to the save path.
        preprocess_radiances(self, files: List[str]) -> xr.Dataset: Preprocesses the radiances from the input files.
        preprocess_cloud_mask(self, files: List[str]) -> xr.Dataset: Preprocesses the cloud mask from the input files.
    """
    @property
    def modis_files(self) -> List[str]:
        """
        Returns a list of all MODIS files in the read path.

        Returns:
            List[str]: A list of file paths.
        """
        # get a list of all MODIS filenames within the path
        files = get_list_filenames(self.read_path, ".hdf")
        return files

    def preprocess_fn(self, ds: xr.Dataset) -> Tuple[xr.Dataset, xr.Dataset]:
        """
        Preprocesses the input dataset by applying corrections etc.

        Args:
            ds (xr.Dataset): The input dataset.

        Returns:
            ds (xr.Dataset):: The preprocessed dataset.
        """
        # copy to avoid modifying original dataset
        ds = ds.copy() 

        # assign coordinate reference system
        ds = add_modis_crs(ds)

        # TODO: Add functionality to resample data to specific resolution

        return ds

    def preprocess_fn_radiances(self, file: List[str]) -> xr.Dataset:
        """
        Preprocesses the MODIS radiance dataset.

        Args:
            file (List[str]): The input file.

        Returns:
            xr.Dataset: The preprocessed dataset.
        """
        # Load file using satpy scenes
        scn = Scene(
            reader="modis_l1b",
            filenames=file
        )
        # Load radiance bands
        channels = get_modis_channel_numbers()
        scn.load(channels, generate=False, calibration='radiance')

        # change to xarray data
        ds = scn.to_xarray_dataset()  

        # do core preprocess function (e.g. resample, add crs etc.)
        ds = self.preprocess_fn(ds) 

        # Store the attributes in a dict before concatenation
        attrs_dict = {x: ds[x].attrs for x in channels}

        # concatinate in new band dimension, and defining a new variable name
        # NOTE: Concatination overwrites attrs of bands.
        ds = ds.assign(Rad=xr.concat(list(map(lambda x: ds[x], channels)), dim="band"))
        # drop duplicate variables
        ds = ds.drop(list(map(lambda x: x, channels)))
        # rename band dimensions
        ds = ds.assign_coords(band=list(map(lambda x: x, channels)))

        # convert measurement time (in seconds) to datetime
        time_stamp = pd.to_datetime(ds.attrs['start_time'])
        time_stamp = time_stamp.strftime("%Y-%m-%d %H:%M")  
        # assign bands and time data to each variable
        ds = ds.assign_coords({"time": [time_stamp]})

        # NOTE: Keep only certain relevant attributes
        ds.attrs = {}
        ds.attrs = dict(
            calibration=attrs_dict[list(attrs_dict.keys())[0]]["calibration"],
            standard_name=attrs_dict[list(attrs_dict.keys())[0]]["standard_name"],
            platform_name=attrs_dict[list(attrs_dict.keys())[0]]["platform_name"],
            sensor=attrs_dict[list(attrs_dict.keys())[0]]["sensor"],
            units=attrs_dict[list(attrs_dict.keys())[0]]["units"],
        )

        # TODO: Correct wavelength assignment. This attaches 36++ wavelengths to each band.
        # assign band wavelengths 
        ds = ds.assign_coords({"band_wavelength": list(MODIS_WAVELENGTHS.values())})   

        return ds

    def preprocess_fn_cloudmask(self, file: List[str]) -> xr.Dataset:
        """
        Preprocesses the input dataset for MODIS cloud masks.

        Args:
            file (List[str]): The input file.

        Returns:
            xr.Dataset: The preprocessed dataset.
        """    
        # Load file using satpy scenes
        scn = Scene(
            reader="modis_l2",
            filenames=file
        )
        # Load cloud mask data
        datasets = scn.available_dataset_names()
        # Needs to be loaded at 1000 m resolution for all channels to match
        scn.load(datasets, generate=False, resolution=1000) 

        # change to xarray data
        ds = scn.to_xarray_dataset()

        return ds

    def preprocess_radiances(self, files: List[str]) -> xr.Dataset:
        """
        Preprocesses radiances from the input files.

        Args:
            files (List[str]): The list of file paths.

        Returns:
            xr.Dataset: The preprocessed dataset.
        """
        identifier = MODIS_NAME_TO_ID[self.satellite]

        # Check that all files contain radiance data
        file = list(filter(lambda x: identifier in x, files))

        # Check that only one file is selected
        logger.info(f"Number of radiance files: {len(file)}")
        assert len(file) == 1

        # load file using satpy, convert to xarray dataset, and preprocess
        ds = self.preprocess_fn_radiances(file)

        return ds


    def preprocess_cloud_mask(self, files: List[str]) -> xr.Dataset:
        """
        Preprocesses cloud mask from the input files.

        Args:
            files (List[str]): The list of file paths.

        Returns:
            xr.Dataset: The preprocessed dataset.
        """
        identifier = MODIS_NAME_TO_ID[f'{self.satellite}_cloud']

        # Check that all files contain radiance data
        file = list(filter(lambda x: identifier in x, files))

        # Check that only one file is selected
        logger.info(f"Number of cloud mask files: {len(file)}")
        assert len(file) == 1

        # load file using satpy, convert to xarray dataset, and preprocess
        ds = self.preprocess_fn_cloudmask(file)

        return ds


    def preprocess_files(self):
        """
        Preprocesses multiple files in read path and saves processed files to save path.
        """
        # get unique times from read path
        unique_times = list(set(map(parse_modis_dates_from_file, self.modis_files)))

        pbar_time = tqdm(unique_times)

        for itime in pbar_time:

            pbar_time.set_description(f"Processing: {itime}")

            # get files from unique times
            files = list(filter(lambda x: itime in x, self.modis_files))

            try:
                # load radiances
                ds = self.preprocess_radiances(files)
            except AssertionError:
                logger.error(f"Skipping {itime} due to error loading")
                continue
            try:
                # load cloud mask
                ds_clouds = self.preprocess_cloud_mask(files)["cloud_mask"]
            except AssertionError:
                logger.error(f"Skipping {itime} due to missing cloud mask")
                continue

            # save cloud mask as data coordinate
            ds = ds.assign_coords({"cloud_mask": (("y", "x"), ds_clouds.values)})
            # add cloud mask attrs to dataset
            ds["cloud_mask"].attrs = ds_clouds.attrs

            # remove crs from dataset
            ds = ds.drop_vars("crs")
            ds = ds.drop_vars("spatial_ref")

            # remove attrs that cause netcdf error
            for attr in ["start_time", "end_time", "area", "_satpy_id"]:
                ds["cloud_mask"].attrs.pop(attr)

            for var in ds.data_vars:
                ds[var].attrs = {}

            # check if save path exists, and create if not
            if not os.path.exists(self.save_path):
                os.makedirs(self.save_path)

            # remove file if it already exists
            itime_name = format_modis_dates(itime)
            save_filename = Path(self.save_path).joinpath(f"{itime_name}_{self.satellite}.nc")
            if os.path.exists(save_filename):
                logger.info(f"File already exists. Overwriting file: {save_filename}")
                os.remove(save_filename)

            # save to netcdf
            ds.to_netcdf(save_filename, engine="netcdf4")

modis_files: List[str] property

Returns a list of all MODIS files in the read path.

Returns:

Type Description
List[str]

List[str]: A list of file paths.

save_path: str instance-attribute

A class for geoprocessing MODIS data.

Attributes:

Name Type Description
satellite str

The satellite to geoprocess data for.
read_path str

The path to read the files from.
save_path str

The path to save the processed files to.

Functions:

Name Description
modis_files

Returns a list of all MODIS files in the read path.
preprocess_fn

xr.Dataset) -> Tuple[xr.Dataset, xr.Dataset]: Preprocesses the input dataset by applying corrections.
preprocess_fn_radiances

xr.Dataset) -> xr.Dataset: Preprocesses the input dataset for radiances.
preprocess_fn_cloudmask

xr.Dataset) -> xr.Dataset: Preprocesses the input dataset for cloud mask.
preprocess_files

Preprocesses the files in the read path and saves the processed files to the save path.
preprocess_radiances

List[str]) -> xr.Dataset: Preprocesses the radiances from the input files.
preprocess_cloud_mask

List[str]) -> xr.Dataset: Preprocesses the cloud mask from the input files.

preprocess_cloud_mask(files)

Preprocesses cloud mask from the input files.

Parameters:

Name Type Description Default
files List[str]

The list of file paths.

 required

Returns:

Type Description
Dataset

xr.Dataset: The preprocessed dataset.
Source code in rs_tools/_src/geoprocessing/modis/geoprocessor_modis.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def preprocess_cloud_mask(self, files: List[str]) -> xr.Dataset:
    """
    Preprocesses cloud mask from the input files.

    Args:
        files (List[str]): The list of file paths.

    Returns:
        xr.Dataset: The preprocessed dataset.
    """
    identifier = MODIS_NAME_TO_ID[f'{self.satellite}_cloud']

    # Check that all files contain radiance data
    file = list(filter(lambda x: identifier in x, files))

    # Check that only one file is selected
    logger.info(f"Number of cloud mask files: {len(file)}")
    assert len(file) == 1

    # load file using satpy, convert to xarray dataset, and preprocess
    ds = self.preprocess_fn_cloudmask(file)

    return ds

preprocess_files()

Preprocesses multiple files in read path and saves processed files to save path.

Source code in rs_tools/_src/geoprocessing/modis/geoprocessor_modis.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def preprocess_files(self):
    """
    Preprocesses multiple files in read path and saves processed files to save path.
    """
    # get unique times from read path
    unique_times = list(set(map(parse_modis_dates_from_file, self.modis_files)))

    pbar_time = tqdm(unique_times)

    for itime in pbar_time:

        pbar_time.set_description(f"Processing: {itime}")

        # get files from unique times
        files = list(filter(lambda x: itime in x, self.modis_files))

        try:
            # load radiances
            ds = self.preprocess_radiances(files)
        except AssertionError:
            logger.error(f"Skipping {itime} due to error loading")
            continue
        try:
            # load cloud mask
            ds_clouds = self.preprocess_cloud_mask(files)["cloud_mask"]
        except AssertionError:
            logger.error(f"Skipping {itime} due to missing cloud mask")
            continue

        # save cloud mask as data coordinate
        ds = ds.assign_coords({"cloud_mask": (("y", "x"), ds_clouds.values)})
        # add cloud mask attrs to dataset
        ds["cloud_mask"].attrs = ds_clouds.attrs

        # remove crs from dataset
        ds = ds.drop_vars("crs")
        ds = ds.drop_vars("spatial_ref")

        # remove attrs that cause netcdf error
        for attr in ["start_time", "end_time", "area", "_satpy_id"]:
            ds["cloud_mask"].attrs.pop(attr)

        for var in ds.data_vars:
            ds[var].attrs = {}

        # check if save path exists, and create if not
        if not os.path.exists(self.save_path):
            os.makedirs(self.save_path)

        # remove file if it already exists
        itime_name = format_modis_dates(itime)
        save_filename = Path(self.save_path).joinpath(f"{itime_name}_{self.satellite}.nc")
        if os.path.exists(save_filename):
            logger.info(f"File already exists. Overwriting file: {save_filename}")
            os.remove(save_filename)

        # save to netcdf
        ds.to_netcdf(save_filename, engine="netcdf4")

preprocess_fn(ds)

Preprocesses the input dataset by applying corrections etc.

Parameters:

Name Type Description Default
ds Dataset

The input dataset.

 required

Returns:

Name Type Description
ds Dataset

: The preprocessed dataset.
Source code in rs_tools/_src/geoprocessing/modis/geoprocessor_modis.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def preprocess_fn(self, ds: xr.Dataset) -> Tuple[xr.Dataset, xr.Dataset]:
    """
    Preprocesses the input dataset by applying corrections etc.

    Args:
        ds (xr.Dataset): The input dataset.

    Returns:
        ds (xr.Dataset):: The preprocessed dataset.
    """
    # copy to avoid modifying original dataset
    ds = ds.copy() 

    # assign coordinate reference system
    ds = add_modis_crs(ds)

    # TODO: Add functionality to resample data to specific resolution

    return ds

preprocess_fn_cloudmask(file)

Preprocesses the input dataset for MODIS cloud masks.

Parameters:

Name Type Description Default
file List[str]

The input file.

 required

Returns:

Type Description
Dataset

xr.Dataset: The preprocessed dataset.
Source code in rs_tools/_src/geoprocessing/modis/geoprocessor_modis.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def preprocess_fn_cloudmask(self, file: List[str]) -> xr.Dataset:
    """
    Preprocesses the input dataset for MODIS cloud masks.

    Args:
        file (List[str]): The input file.

    Returns:
        xr.Dataset: The preprocessed dataset.
    """    
    # Load file using satpy scenes
    scn = Scene(
        reader="modis_l2",
        filenames=file
    )
    # Load cloud mask data
    datasets = scn.available_dataset_names()
    # Needs to be loaded at 1000 m resolution for all channels to match
    scn.load(datasets, generate=False, resolution=1000) 

    # change to xarray data
    ds = scn.to_xarray_dataset()

    return ds

preprocess_fn_radiances(file)

Preprocesses the MODIS radiance dataset.

Parameters:

Name Type Description Default
file List[str]

The input file.

 required

Returns:

Type Description
Dataset

xr.Dataset: The preprocessed dataset.
Source code in rs_tools/_src/geoprocessing/modis/geoprocessor_modis.py 
 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
def preprocess_fn_radiances(self, file: List[str]) -> xr.Dataset:
    """
    Preprocesses the MODIS radiance dataset.

    Args:
        file (List[str]): The input file.

    Returns:
        xr.Dataset: The preprocessed dataset.
    """
    # Load file using satpy scenes
    scn = Scene(
        reader="modis_l1b",
        filenames=file
    )
    # Load radiance bands
    channels = get_modis_channel_numbers()
    scn.load(channels, generate=False, calibration='radiance')

    # change to xarray data
    ds = scn.to_xarray_dataset()  

    # do core preprocess function (e.g. resample, add crs etc.)
    ds = self.preprocess_fn(ds) 

    # Store the attributes in a dict before concatenation
    attrs_dict = {x: ds[x].attrs for x in channels}

    # concatinate in new band dimension, and defining a new variable name
    # NOTE: Concatination overwrites attrs of bands.
    ds = ds.assign(Rad=xr.concat(list(map(lambda x: ds[x], channels)), dim="band"))
    # drop duplicate variables
    ds = ds.drop(list(map(lambda x: x, channels)))
    # rename band dimensions
    ds = ds.assign_coords(band=list(map(lambda x: x, channels)))

    # convert measurement time (in seconds) to datetime
    time_stamp = pd.to_datetime(ds.attrs['start_time'])
    time_stamp = time_stamp.strftime("%Y-%m-%d %H:%M")  
    # assign bands and time data to each variable
    ds = ds.assign_coords({"time": [time_stamp]})

    # NOTE: Keep only certain relevant attributes
    ds.attrs = {}
    ds.attrs = dict(
        calibration=attrs_dict[list(attrs_dict.keys())[0]]["calibration"],
        standard_name=attrs_dict[list(attrs_dict.keys())[0]]["standard_name"],
        platform_name=attrs_dict[list(attrs_dict.keys())[0]]["platform_name"],
        sensor=attrs_dict[list(attrs_dict.keys())[0]]["sensor"],
        units=attrs_dict[list(attrs_dict.keys())[0]]["units"],
    )

    # TODO: Correct wavelength assignment. This attaches 36++ wavelengths to each band.
    # assign band wavelengths 
    ds = ds.assign_coords({"band_wavelength": list(MODIS_WAVELENGTHS.values())})   

    return ds

preprocess_radiances(files)

Preprocesses radiances from the input files.

Parameters:

Name Type Description Default
files List[str]

The list of file paths.

 required

Returns:

Type Description
Dataset

xr.Dataset: The preprocessed dataset.
Source code in rs_tools/_src/geoprocessing/modis/geoprocessor_modis.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def preprocess_radiances(self, files: List[str]) -> xr.Dataset:
    """
    Preprocesses radiances from the input files.

    Args:
        files (List[str]): The list of file paths.

    Returns:
        xr.Dataset: The preprocessed dataset.
    """
    identifier = MODIS_NAME_TO_ID[self.satellite]

    # Check that all files contain radiance data
    file = list(filter(lambda x: identifier in x, files))

    # Check that only one file is selected
    logger.info(f"Number of radiance files: {len(file)}")
    assert len(file) == 1

    # load file using satpy, convert to xarray dataset, and preprocess
    ds = self.preprocess_fn_radiances(file)

    return ds

geoprocess(satellite, read_path='./', save_path='./')

Geoprocesses MODIS files

Parameters:

Name Type Description Default
satellite str

The satellite of the data to geoprocess.

 required
read_path str

The path to read the files from. Defaults to "./".

 './'
save_path str

The path to save the geoprocessed files to. Defaults to "./".

 './'

Returns:

Type Description

None
Source code in rs_tools/_src/geoprocessing/modis/geoprocessor_modis.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
def geoprocess(
        satellite: str,
        read_path: str = "./",
        save_path: str = "./"
):
    """
    Geoprocesses MODIS files

    Args:
        satellite (str, optional): The satellite of the data to geoprocess.
        read_path (str, optional): The path to read the files from. Defaults to "./".
        save_path (str, optional): The path to save the geoprocessed files to. Defaults to "./".

    Returns:
        None
    """
    # Initialize MODIS GeoProcessor
    logger.info(f"Initializing {satellite.upper()} GeoProcessor...")
    logger.debug(f"PATHH: {read_path}")
    modis_geoprocessor = MODISGeoProcessing(
        satellite=satellite, 
        read_path=read_path, 
        save_path=save_path
        )
    logger.info(f"GeoProcessing Files...")
    modis_geoprocessor.preprocess_files()

    logger.info(f"Finished {satellite.upper()} GeoProcessing Script...!")