Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self._init_refl3x(projectables)
# Derive the sun-zenith angles, and use the nir and thermal ir
# brightness tempertures and derive the reflectance using
# PySpectral. The reflectance is stored internally in PySpectral and
# needs to be derived first in order to get the emissive part.
_ = self._get_reflectance(projectables, optional_datasets)
_nir, _ = projectables
proj = Dataset(self._refl3x.emissive_part_3x(), **_nir.attrs)
proj.attrs['units'] = 'K'
self.apply_modifier_info(_nir, proj)
return proj
class PSPAtmosphericalCorrection(CompositeBase):
def __call__(self, projectables, optional_datasets=None, **info):
"""Get the atmospherical correction. Uses pyspectral.
"""
from pyspectral.atm_correction_ir import AtmosphericalCorrection
band = projectables[0]
if optional_datasets:
satz = optional_datasets[0]
else:
from pyorbital.orbital import get_observer_look
lons, lats = band.attrs['area'].get_lonlats_dask(CHUNK_SIZE)
try:
dummy, satel = get_observer_look(band.attrs['satellite_longitude'],
correction_limit (float): Maximum solar zenith angle to apply the
correction in degrees. Pixels beyond this limit have a
constant correction applied. Default 88.
max_sza (float): Maximum solar zenith angle in degrees that is
considered valid and correctable. Default 95.0.
"""
self.correction_limit = correction_limit
super(EffectiveSolarPathLengthCorrector, self).__init__(**kwargs)
def _apply_correction(self, proj, coszen):
LOG.debug("Apply the effective solar atmospheric path length correction method by Li and Shibata")
return atmospheric_path_length_correction(proj, coszen, limit=self.correction_limit, max_sza=self.max_sza)
class PSPRayleighReflectance(CompositeBase):
"""Pyspectral-based rayleigh corrector for visible channels."""
_rayleigh_cache = WeakValueDictionary()
def get_angles(self, vis):
"""Get the sun and satellite angles from the current dataarray."""
from pyorbital.astronomy import get_alt_az, sun_zenith_angle
from pyorbital.orbital import get_observer_look
lons, lats = vis.attrs['area'].get_lonlats(chunks=vis.data.chunks)
sunalt, suna = get_alt_az(vis.attrs['start_time'], lons, lats)
suna = np.rad2deg(suna)
sunz = sun_zenith_angle(vis.attrs['start_time'], lons, lats)
sat_lon, sat_lat, sat_alt = get_satpos(vis)
sata, satel = get_observer_look(
return sunzen_corr_cos(proj, coszen)
class EffectiveSolarPathLengthCorrector(SunZenithCorrectorBase):
"""Special sun zenith correction with the method proposed by Li and Shibata
(2006): https://doi.org/10.1175/JAS3682.1
"""
def _apply_correction(self, proj, coszen):
LOG.debug(
"Apply the effective solar atmospheric path length correction method by Li and Shibata")
return atmospheric_path_length_correction(proj, coszen)
class SMACReflectance(CompositeBase):
def __call__(self, projectables, optional_datasets=None, **info):
from smac import smac_inv, PdeZ, coeff
(vis, ) = projectables
nom_smac = '/home/a001673/usr/contrib/smac-python/COEFS/coef_MODIS%s_CONT.dat' % vis.attrs['name']
coefs = coeff(nom_smac)
(phi_v, theta_v, phi_s, theta_s) = optional_datasets
# theta_s=30 #solar zenith angle
# phi_s=180 #solar azimuth angle
# theta_v=0 #viewing zenith angle
# phi_v=0 #viewing azimuth
# compute pressure at pixel altitude (in m)
pressure = PdeZ(0)
# find the values of AOT, UO3, UH2O
AOT550 = 0.1 # AOT at 550 nm
# Derive the sun-zenith angles, and use the nir and thermal ir
# brightness tempertures and derive the reflectance using
# PySpectral. The reflectance is stored internally in PySpectral and
# needs to be derived first in order to get the emissive part.
_ = self._get_reflectance(projectables, optional_datasets)
_nir, _ = projectables
proj = xr.DataArray(self._refl3x.emissive_part_3x(), attrs=_nir.attrs,
dims=_nir.dims, coords=_nir.coords)
proj.attrs['units'] = 'K'
self.apply_modifier_info(_nir, proj)
return proj
class PSPAtmosphericalCorrection(CompositeBase):
"""Correct for atmospheric effects."""
def __call__(self, projectables, optional_datasets=None, **info):
"""Get the atmospherical correction.
Uses pyspectral.
"""
from pyspectral.atm_correction_ir import AtmosphericalCorrection
band = projectables[0]
if optional_datasets:
satz = optional_datasets[0]
else:
from pyorbital.orbital import get_observer_look
lons, lats = band.attrs['area'].get_lonlats(chunks=CHUNK_SIZE)
LOG.warning("Could not get the reflectance correction using band name: %s", vis.attrs['name'])
LOG.warning("Will try use the wavelength, however, this may be ambiguous!")
refl_cor_band = corrector.get_reflectance(sunz.load().values,
satz.load().values,
ssadiff.load().values,
vis.attrs['wavelength'][1],
red.values)
proj = vis - refl_cor_band
proj = proj.clip(0, 100)
proj.attrs = vis.attrs
self.apply_modifier_info(vis, proj)
return proj
class NIRReflectance(CompositeBase):
def __call__(self, projectables, optional_datasets=None, **info):
"""Get the reflectance part of an NIR channel. Not supposed to be used
for wavelength outside [3, 4] µm.
"""
self._init_refl3x(projectables)
_nir, _ = projectables
proj = Dataset(self._get_reflectance(projectables, optional_datasets) * 100, **_nir.info)
proj.info['units'] = '%'
self.apply_modifier_info(_nir, proj)
return proj
def _init_refl3x(self, projectables):
"""Initiate the 3.x reflectance derivations
data = projectables[0]
new_attrs = data.attrs.copy()
new_attrs.update({key: val
for (key, val) in attrs.items()
if val is not None})
resolution = new_attrs.get('resolution', None)
new_attrs.update(self.attrs)
if resolution is not None:
new_attrs['resolution'] = resolution
return xr.DataArray(data=data.data, attrs=new_attrs,
dims=data.dims, coords=data.coords)
class GenericCompositor(CompositeBase):
"""Basic colored composite builder."""
modes = {1: 'L', 2: 'LA', 3: 'RGB', 4: 'RGBA'}
def __init__(self, name, common_channel_mask=True, **kwargs):
"""Collect custom configuration values.
Args:
common_channel_mask (bool): If True, mask all the channels with
a mask that combines all the invalid areas of the given data.
"""
self.common_channel_mask = common_channel_mask
super(GenericCompositor, self).__init__(name, **kwargs)
@classmethod
UO3 = 0.3 # Ozone content (cm) 0.3 cm= 300 Dobson Units
UH2O = 3 # Water vapour (g/cm2)
# compute the atmospheric correction
LOG.info('Using SMAC')
res = smac_inv(vis.values / 100.0, theta_s, phi_s, theta_v, phi_v, pressure,
AOT550, UO3, UH2O, coefs)
res *= 100
res = res.clip(0, 100)
res.attrs = vis.attrs.copy()
self.apply_modifier_info(vis, res)
return res
class PSPRayleighReflectance(CompositeBase):
def __call__(self, projectables, optional_datasets=None, **info):
"""Get the corrected reflectance when removing Rayleigh scattering.
Uses pyspectral.
"""
from pyspectral.rayleigh import Rayleigh
(vis, red) = projectables
if vis.shape != red.shape:
raise IncompatibleAreas
try:
(sata, satz, suna, sunz) = optional_datasets
except ValueError:
from pyorbital.astronomy import get_alt_az, sun_zenith_angle
from pyorbital.orbital import get_observer_look
elif any(a is None for a in areas):
raise ValueError("Missing 'area' attribute")
if not all(areas[0] == x for x in areas[1:]):
LOG.debug("Not all areas are the same in "
"'{}'".format(self.attrs['name']))
raise IncompatibleAreas("Areas are different")
def check_areas(self, data_arrays):
"""Check that the areas of the *data_arrays* are compatible."""
warnings.warn('satpy.composites.CompositeBase.check_areas is deprecated, use '
'satpy.composites.CompositeBase.match_data_arrays instead')
return self.match_data_arrays(data_arrays)
class SunZenithCorrectorBase(CompositeBase):
"""Base class for sun zenith correction."""
coszen = WeakValueDictionary()
def __init__(self, max_sza=95.0, **kwargs):
"""Collect custom configuration values.
Args:
max_sza (float): Maximum solar zenith angle in degrees that is
considered valid and correctable. Default 95.0.
"""
self.max_sza = max_sza
self.max_sza_cos = np.cos(np.deg2rad(max_sza)) if max_sza is not None else None
super(SunZenithCorrectorBase, self).__init__(**kwargs)
"""
(ir_039, ir_108, ir_134) = projectables
LOG.info('Applying CO2 correction')
dt_co2 = (ir_108 - ir_134) / 4.0
rcorr = ir_108**4 - (ir_108 - dt_co2)**4
t4_co2corr = (ir_039**4 + rcorr).clip(0.0) ** 0.25
t4_co2corr.attrs = ir_039.attrs.copy()
self.apply_modifier_info(ir_039, t4_co2corr)
return t4_co2corr
class DifferenceCompositor(CompositeBase):
"""Make the difference of two data arrays."""
def __call__(self, projectables, nonprojectables=None, **info):
"""Generate the composite."""
if len(projectables) != 2:
raise ValueError("Expected 2 datasets, got %d" % (len(projectables),))
projectables = self.match_data_arrays(projectables)
info = combine_metadata(*projectables)
info['name'] = self.attrs['name']
proj = projectables[0] - projectables[1]
proj.attrs = info
return proj
class SingleBandCompositor(CompositeBase):
return t4_co2corr
class DifferenceCompositor(CompositeBase):
def __call__(self, projectables, nonprojectables=None, **info):
if len(projectables) != 2:
raise ValueError("Expected 2 datasets, got %d" %
(len(projectables), ))
info = combine_metadata(*projectables)
info['name'] = self.attrs['name']
return Dataset(projectables[0] - projectables[1], **info)
class GenericCompositor(CompositeBase):
modes = {1: 'L', 2: 'LA', 3: 'RGB', 4: 'RGBA'}
def check_area_compatibility(self, projectables):
areas = [projectable.attrs.get('area', None)
for projectable in projectables]
areas = [area for area in areas if area is not None]
if areas and areas.count(areas[0]) != len(areas):
LOG.debug("Not all areas are the same in '{}'".format(self.attrs['name']))
raise IncompatibleAreas
def _concat_datasets(self, projectables, mode):
self.check_area_compatibility(projectables)
try:
data = xr.concat(projectables, 'bands', coords='minimal')