Source code for satpy.readers.vii_base_nc

#!/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
# 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 <>.

"""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] @staticmethod def _perform_interpolation(variable): """Perform the interpolation from tie points to pixel points. Args: variable: xarray DataArray containing the dataset to interpolate. Returns: DataArray: array containing the interpolate values, all the original metadata and the updated dimension names. """ interpolated_values = tie_points_interpolation( [variable], SCAN_ALT_TIE_POINTS, TIE_POINTS_FACTOR )[0] new_variable = interpolated_values.rename( num_tie_points_act='num_pixels', num_tie_points_alt='num_lines' ) = new_variable.attrs = variable.attrs return new_variable
[docs] @staticmethod def _perform_geo_interpolation(longitude, latitude): """Perform the interpolation of geographic coodinates from tie points to pixel points. Args: longitude: xarray DataArray containing the longitude dataset to interpolate. latitude: xarray DataArray containing the longitude dataset to interpolate. Returns: tuple of arrays containing the interpolate values, all the original metadata and the updated dimension names. """ interpolated_longitude, interpolated_latitude = tie_points_geo_interpolation( longitude, latitude, SCAN_ALT_TIE_POINTS, TIE_POINTS_FACTOR ) new_longitude = interpolated_longitude.rename( num_tie_points_act='num_pixels', num_tie_points_alt='num_lines' ) = new_longitude.attrs = longitude.attrs new_latitude = interpolated_latitude.rename( num_tie_points_act='num_pixels', num_tie_points_alt='num_lines' ) = new_latitude.attrs = latitude.attrs return new_longitude, new_latitude
[docs] def _perform_orthorectification(self, variable, orthorect_data_name): """Perform the orthorectification.""" raise NotImplementedError
[docs] def _perform_calibration(self, variable, dataset_info): """Perform the calibration.""" raise NotImplementedError
[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