Source code for satpy.readers.viirs_edr_flood

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2019 Satpy developers
#
# This file is part of satpy.
#
# satpy is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# satpy is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# satpy.  If not, see <http://www.gnu.org/licenses/>.
"""Interface to VIIRS flood product."""

import numpy as np
from pyresample import geometry

from satpy.readers.hdf4_utils import HDF4FileHandler




class VIIRSEDRFlood(HDF4FileHandler):
    """VIIRS EDR Flood-product handler for HDF4 files."""

    @property
    def start_time(self):
        """Get start time."""
        return self.filename_info["start_time"]

    @property
    def end_time(self):
        """Get end time."""
        return self.filename_info.get("end_time", self.start_time)

    @property
    def sensor_name(self):
        """Get sensor name."""
        sensor = self["/attr/SensorIdentifyCode"]
        if isinstance(sensor, np.ndarray):
            return str(sensor.astype(str)).lower()
        return sensor.lower()

    @property
    def platform_name(self):
        """Get platform name."""
        platform_name = self["/attr/Satellitename"]
        if isinstance(platform_name, np.ndarray):
            return str(platform_name.astype(str)).lower()
        return platform_name.lower()



    def get_metadata(self, data, ds_info):
        """Get metadata."""
        metadata = {}
        metadata.update(data.attrs)
        metadata.update(ds_info)
        metadata.update({
            "sensor": self.sensor_name,
            "platform_name": self.platform_name,
            "start_time": self.start_time,
            "end_time": self.end_time,
        })

        return metadata




    def get_dataset(self, ds_id, ds_info):
        """Get dataset."""
        data = self[ds_id["name"]]

        data.attrs = self.get_metadata(data, ds_info)

        fill = data.attrs.pop("_Fillvalue")
        offset = data.attrs.get("add_offset")
        scale_factor = data.attrs.get("scale_factor")

        data = data.where(data != fill)
        if scale_factor is not None and offset is not None:
            data *= scale_factor
            data += offset

        return data




    def get_area_def(self, ds_id):
        """Get area definition."""
        data = self[ds_id["name"]]

        proj_dict = {
            "proj": "latlong",
            "datum": "WGS84",
            "ellps": "WGS84",
            "no_defs": True
        }

        area_extent = [data.attrs.get("ProjectionMinLongitude"), data.attrs.get("ProjectionMinLatitude"),
                       data.attrs.get("ProjectionMaxLongitude"), data.attrs.get("ProjectionMaxLatitude")]

        area = geometry.AreaDefinition(
            "viirs_flood_area",
            "name_of_proj",
            "id_of_proj",
            proj_dict,
            int(self.filename_info["dim0"]),
            int(self.filename_info["dim1"]),
            np.asarray(area_extent)
        )

        return area