#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (c) 2020 Satpy developers
#
# 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/>.
"""EUMETSAT EPS-SG Visible/Infrared Imager (VII) readers base class."""
import logging
from datetime import datetime
from geotiepoints.viiinterpolator import tie_points_geo_interpolation, tie_points_interpolation
from satpy.readers.netcdf_utils import NetCDF4FileHandler
from satpy.readers.vii_utils import SCAN_ALT_TIE_POINTS, TIE_POINTS_FACTOR
logger = logging.getLogger(__name__)
[docs]
class ViiNCBaseFileHandler(NetCDF4FileHandler):
"""Base reader class for VII products in netCDF format.
Args:
filename (str): File to read
filename_info (dict): Dictionary with filename information
filetype_info (dict): Dictionary with filetype information
orthorect (bool): activates the orthorectification correction where available
"""
def __init__(self, filename, filename_info, filetype_info, orthorect=False):
"""Prepare the class for dataset reading."""
super().__init__(filename, filename_info, filetype_info, auto_maskandscale=True)
# Saves the orthorectification flag
self.orthorect = orthorect and filetype_info.get("orthorect", True)
# Saves the interpolation flag
self.interpolate = filetype_info.get("interpolate", True)
try:
longitude = self[filetype_info["cached_longitude"]]
latitude = self[filetype_info["cached_latitude"]]
if self.interpolate:
self.longitude, self.latitude = self._perform_geo_interpolation(longitude, latitude)
else:
self.longitude, self.latitude = longitude, latitude
except KeyError:
logger.warning("Cached longitude and/or latitude datasets are not correctly defined in YAML file")
self.longitude, self.latitude = None, None
[docs]
def _standardize_dims(self, variable):
"""Standardize dims to y, x."""
if "num_pixels" in variable.dims:
variable = variable.rename({"num_pixels": "x", "num_lines": "y"})
if "num_points_act" in variable.dims:
variable = variable.rename({"num_points_act": "x", "num_points_alt": "y"})
if variable.dims[0] == "x":
variable = variable.transpose("y", "x")
return variable
[docs]
def get_dataset(self, dataset_id, dataset_info):
"""Get dataset using file_key in dataset_info."""
var_key = dataset_info["file_key"]
logger.debug("Reading in file to get dataset with key %s.", var_key)
if var_key == "cached_longitude" and self.longitude is not None:
variable = self.longitude.copy()
elif var_key == "cached_latitude" and self.latitude is not None:
variable = self.latitude.copy()
else:
try:
variable = self[var_key]
except KeyError:
logger.warning("Could not find key %s in NetCDF file, no valid Dataset created", var_key)
return None
# If the dataset is marked for interpolation, perform the interpolation from tie points to pixels
if dataset_info.get("interpolate", False) and self.interpolate:
variable = self._perform_interpolation(variable)
# Perform the calibration if required
if dataset_info.get("calibration") is not None:
variable = self._perform_calibration(variable, dataset_info)
# Perform the orthorectification if required
if self.orthorect:
orthorect_data_name = dataset_info.get("orthorect_data", None)
if orthorect_data_name is not None:
variable = self._perform_orthorectification(variable, orthorect_data_name)
# Manage the attributes of the dataset
variable.attrs.setdefault("units", None)
variable.attrs.update(dataset_info)
variable.attrs.update(self._get_global_attributes())
variable = self._standardize_dims(variable)
return variable
[docs]
def _get_global_attributes(self):
"""Create a dictionary of global attributes to be added to all datasets."""
attributes = {
"filename": self.filename,
"start_time": self.start_time,
"end_time": self.end_time,
"spacecraft_name": self.spacecraft_name,
"ssp_lon": self.ssp_lon,
"sensor": self.sensor,
"filename_start_time": self.filename_info["sensing_start_time"],
"filename_end_time": self.filename_info["sensing_end_time"],
"platform_name": self.spacecraft_name,
}
# Add a "quality_group" item to the dictionary with all the variables and attributes
# which are found in the 'quality' group of the VII product
quality_group = self["quality"]
quality_dict = {}
for key in quality_group:
# Add the values (as Numpy array) of each variable in the group where possible
try:
quality_dict[key] = quality_group[key].values
except ValueError:
quality_dict[key] = None
# Add the attributes of the quality group
quality_dict.update(quality_group.attrs)
attributes["quality_group"] = quality_dict
return attributes
@property
def start_time(self):
"""Get observation start time."""
try:
start_time = datetime.strptime(self["/attr/sensing_start_time_utc"], "%Y%m%d%H%M%S.%f")
except ValueError:
start_time = datetime.strptime(self["/attr/sensing_start_time_utc"], "%Y-%m-%d %H:%M:%S.%f")
return start_time
@property
def end_time(self):
"""Get observation end time."""
try:
end_time = datetime.strptime(self["/attr/sensing_end_time_utc"], "%Y%m%d%H%M%S.%f")
except ValueError:
end_time = datetime.strptime(self["/attr/sensing_end_time_utc"], "%Y-%m-%d %H:%M:%S.%f")
return end_time
@property
def spacecraft_name(self):
"""Return spacecraft name."""
return self["/attr/spacecraft"]
@property
def sensor(self):
"""Return sensor."""
return self["/attr/instrument"]
@property
def ssp_lon(self):
"""Return subsatellite point longitude."""
# This parameter is not applicable to VII
return None