Source code for satpy.readers.hy2_scat_l2b_h5

# -*- coding: utf-8 -*-
# Copyright (c) 2020,2021 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/>.

"""HY-2B L2B Reader.

Distributed by Eumetsat in HDF5 format.
Also handle the HDF5 files from NSOAS, based on a file example.
"""

from datetime import datetime

import numpy as np
import xarray as xr

from satpy.readers.hdf5_utils import HDF5FileHandler




class HY2SCATL2BH5FileHandler(HDF5FileHandler):
    """File handler for HY2 scat."""

    @property
    def start_time(self):
        """Time for first observation."""
        return datetime.strptime(self["/attr/Range_Beginning_Time"],
                                 "%Y%m%dT%H:%M:%S")

    @property
    def end_time(self):
        """Time for final observation."""
        return datetime.strptime(self["/attr/Range_Ending_Time"],
                                 "%Y%m%dT%H:%M:%S")

    @property
    def platform_name(self):
        """Get the Platform ShortName."""
        return self["/attr/Platform_ShortName"]



    def get_variable_metadata(self):
        """Get the variable metadata."""
        info = getattr(self, "attrs", {})
        info.update({
            "Equator_Crossing_Longitude": self["/attr/Equator_Crossing_Longitude"],
            "Equator_Crossing_Time": self["/attr/Equator_Crossing_Time"],
            "Input_L2A_Filename": self["/attr/Input_L2A_Filename"],
            "L2B_Actual_WVC_Rows": self["/attr/L2B_Actual_WVC_Rows"],
            "Orbit_Inclination": self["/attr/Orbit_Inclination"],
            "Orbit_Number": self["/attr/Orbit_Number"],
            "Output_L2B_Filename": self["/attr/Output_L2B_Filename"],
            "Production_Date_Time": self["/attr/Production_Date_Time"],
            "L2B_Expected_WVC_Rows": self["/attr/L2B_Expected_WVC_Rows"]
        })
        try:
            info.update({"L2B_Number_WVC_cells": self["/attr/L2B_Number_WVC_cells"]})
        except KeyError:
            info.update({"L2B_Expected_WVC_Cells": self["/attr/L2B_Expected_WVC_Cells"]})
        return info




    def get_metadata(self):
        """Get the metadata."""
        info = getattr(self, "attrs", {})
        info.update({
            "WVC_Size": self["/attr/WVC_Size"],
            "HDF_Version_Id": self["/attr/HDF_Version_Id"],
            "Instrument_ShorName": self["/attr/Instrument_ShorName"],
            "L2A_Inputdata_Version": self["/attr/L2A_Inputdata_Version"],
            "L2B_Algorithm_Descriptor": self["/attr/L2B_Algorithm_Descriptor"],
            "L2B_Data_Version": self["/attr/L2B_Data_Version"],
            "L2B_Processing_Type": self["/attr/L2B_Processing_Type"],
            "L2B_Processor_Name": self["/attr/L2B_Processor_Name"],
            "L2B_Processor_Version": self["/attr/L2B_Processor_Version"],
            "Long_Name": self["/attr/Long_Name"],
            "Platform_LongName": self["/attr/Platform_LongName"],
            "Platform_ShortName": self["/attr/Platform_ShortName"],
            "Platform_Type": self["/attr/Platform_Type"],
            "Producer_Agency": self["/attr/Producer_Agency"],
            "Producer_Institution": self["/attr/Producer_Institution"],
            "Rev_Orbit_Perio": self["/attr/Rev_Orbit_Period"],
            "Short_Name": self["/attr/Short_Name"],
            "Sigma0_Granularity": self["/attr/Sigma0_Granularity"],
        })
        return info




    def get_dataset(self, key, info):
        """Get the dataset."""
        dims = ["y", "x"]
        if self[key["name"]].ndim == 3:
            dims = ["y", "x", "selection"]
        data = self[key["name"]]
        if "valid range" in data.attrs:
            data.attrs.update({"valid_range": data.attrs.pop("valid range")})
        if key["name"] in "wvc_row_time":
            data = data.rename({data.dims[0]: "y"})
        else:
            dim_map = {curr_dim: new_dim for curr_dim, new_dim in zip(data.dims, dims)}
            data = data.rename(dim_map)
            data = self._mask_data(data)
            data = self._scale_data(data)

            if key["name"] in "wvc_lon":
                _attrs = data.attrs
                data = xr.where(data > 180, data - 360., data)
                data.attrs.update(_attrs)
        data.attrs.update(info)
        data.attrs.update(self.get_metadata())
        data.attrs.update(self.get_variable_metadata())
        if "Platform_ShortName" in data.attrs:
            data.attrs.update({"platform_name": data.attrs["Platform_ShortName"]})

        return data




    def _scale_data(self, data):
        return data * data.attrs["scale_factor"] + data.attrs["add_offset"]




    def _mask_data(self, data):
        _attrs = data.attrs
        valid_range = data.attrs["valid_range"]
        data = xr.where(data == data.attrs["fill_value"], np.nan, data)
        data = xr.where(data < valid_range[0], np.nan, data)
        data = xr.where(data > valid_range[1], np.nan, data)
        data.attrs.update(_attrs)
        return data