#! /usr/bin/env python3
#
# Copyright 2018 California Institute of Technology
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# ISOFIT: Imaging Spectrometer Optimal FITting
# Author: Philip G. Brodrick, philip.brodrick@jpl.nasa.gov
import logging
import os
from collections import OrderedDict
from typing import Dict, List, Type
import numpy as np
from isofit.configs.base_config import BaseConfigSection
from isofit.configs.sections.statevector_config import StateVectorConfig
from isofit.data import env
[docs]
class RadiativeTransferEngineConfig(BaseConfigSection):
"""
Radiative transfer unknowns configuration.
"""
def __init__(self, sub_configdic: dict = None, name: str = None):
super().__init__()
"""str: Name of config - optional, and not currently used."""
[docs]
self._engine_name_type = str
[docs]
self.engine_name = None
"""str: Name of radiative transfer engine to use - options ['modtran', '6s', 'sRTMnet']."""
[docs]
self._engine_base_dir_type = str
[docs]
self.engine_base_dir = None
"""str: base directory of the given radiative transfer engine on user's OS."""
[docs]
self._engine_lut_file_type = str
[docs]
self.engine_lut_file = None
"""str: File containing the prebuilt LUT file (hdf5)."""
[docs]
self._wavelength_file_type = str
[docs]
self.wavelength_file = None
"""str: Optional path to wavelength file for high-res atmospheric calculations"""
[docs]
self._wavelength_range_type = list()
[docs]
self.wavelength_range = None
"""List: The wavelength range to execute this radiative transfer engine over."""
[docs]
self._environment_type = str
[docs]
self.environment = None
"""str: Additional environment directives for the shell script."""
[docs]
self._lut_path_type = str
"""str: The path to the look up table directory used by the radiative transfer engine."""
[docs]
self._sim_path_type = str
"""str: Path to the simulation outputs for the radiative transfer engine."""
[docs]
self._template_file_type = str
[docs]
self.template_file = None
"""str: A template file to be used as the base-configuration for the given radiative transfer engine."""
[docs]
self._treat_as_emissive_type = bool
[docs]
self.treat_as_emissive = False
"""bool: Run the simulation in emission mode"""
[docs]
self._glint_model_type = bool
[docs]
self.glint_model = False
"""
Flag to indicate whether to use the sun and sky glint model from Gege (2012, 2014) in the forward model.
Only currently functional with multipart MODTRAN.
"""
[docs]
self._rt_mode_type = str
[docs]
self.rt_mode = "transm"
"""str: Radiative transfer mode of LUT simulations.
'transm' for transmittances, 'rdn' for reflected radiance."""
[docs]
self._lut_names_type = dict
"""Dictionary: Names of the elements to run this radiative transfer element on. Must be a subset
of the keys in radiative_transfer->lut_grid. If not specified, uses all keys from
radiative_transfer-> lut_grid. Auto-sorted (alphabetically) below."""
[docs]
self._statevector_names_type = list()
[docs]
self.statevector_names = None
"""List: Names of the statevector elements to use with this radiative transfer engine. Must be a subset
of the keys in radiative_transfer->statevector. If not specified, uses all keys from
radiative_transfer->statevector. Auto-sorted (alphabetically) below."""
[docs]
self._lut_compression_type = str
[docs]
self.lut_compression = "zlib"
"""str: Compression method to use for the LUT NetCDF"""
[docs]
self._lut_complevel_type = int
[docs]
self.lut_complevel = None
"""int: The compression level to use for the chosen method"""
# MODTRAN parameters
[docs]
self._aerosol_template_file_type = str
[docs]
self.aerosol_template_file = None
"""str: Aerosol template file, currently only implemented for MODTRAN."""
[docs]
self._aerosol_model_file_type = str
[docs]
self.aerosol_model_file = None
"""str: Aerosol model file, currently only implemented for MODTRAN."""
[docs]
self._multipart_transmittance_type = bool
[docs]
self.multipart_transmittance = False
"""str: Use True to specify triple-run diffuse & direct transmittance
estimation. Only implemented for MODTRAN."""
# MODTRAN simulator
[docs]
self._emulator_file_type = str
[docs]
self.emulator_file = None
"""str: Path to emulator model file"""
[docs]
self._emulator_aux_file_type = str
[docs]
self.emulator_aux_file = None
"""str: path to emulator auxiliary data - expected npz format"""
[docs]
self._parallel_layer_read_type = bool
[docs]
self.parallel_layer_read = True
"""bool: Flag for how to load and run sRTMnet prediction.
If True, will read in the weights/biases per layer per worker.
If False, will load entire model into shared memory
Model doesn't always fit in shared memory for smaller systems
"""
[docs]
self._predict_parallel_chunks_type = int
[docs]
self.predict_parallel_chunks = 20
"""int: If emulator predictions are in parallel. How many chunks to run.
Keep to a small number for systems with little memory and slow read.
"""
# 6S parameters - not the corcommemnd
# TODO: these should come from a template file, as in modtran
"""int: 6s-only day parameter."""
"""int: 6s-only month parameter."""
[docs]
self._elev_type = float
"""float: 6s-only elevation parameter."""
"""float: 6s-only altitude parameter."""
[docs]
self._obs_file_type = str
"""str: 6s-only observation file."""
[docs]
self._solzen_type = float
"""float: 6s-only solar zenith."""
[docs]
self._solaz_type = float
"""float: 6s-only solar azimuth."""
[docs]
self._viewzen_type = float
"""float: 6s-only view zenith."""
[docs]
self._viewaz_type = float
"""float: 6s-only view azimuth."""
[docs]
self._earth_sun_distance_file_type = str
[docs]
self.earth_sun_distance_file = ""
"""str: 6s-only earth-to-sun distance file."""
[docs]
self._irradiance_file_type = str
[docs]
self.irradiance_file = ""
"""str: 6s-only irradiance file."""
# MODTRAN and 6S
"""bool: Indicates that code should terminate as soon as all radiative transfer engine configuration files are
written (without running them)"""
self.set_config_options(sub_configdic)
if self.lut_names is not None:
keys = list(self.lut_names.keys())
keys.sort()
self.lut_names = {i: self.lut_names[i] for i in keys}
if self.statevector_names is not None:
self.statevector_names.sort()
[docs]
def _check_config_validity(self) -> List[str]:
errors = list()
valid_rt_engines = ["modtran", "6s", "sRTMnet", "KernelFlowsGP"]
if self.engine_name not in valid_rt_engines:
errors.append(
"radiative_transfer->raditive_transfer_model: {} not in one of the"
" available models: {}".format(self.engine_name, valid_rt_engines)
)
valid_rt_modes = ["transm", "rdn"]
if self.rt_mode not in valid_rt_modes:
errors.append(
"radiative_transfer->raditive_transfer_mode: {} not in one of the"
" available modes: {}".format(self.rt_mode, valid_rt_modes)
)
# Only check for missing files when a prebuilt LUT is not provided
if not os.path.exists(self.lut_path):
# Check that all input files exist
for key in self._get_nontype_attributes():
value = getattr(self, key)
if value and key[-5:] == "_file" and key != "emulator_file":
if os.path.isfile(value) is False:
errors.append(
"Config value radiative_transfer->{}: {} not found".format(
key, value
)
)
if self.earth_sun_distance_file is None and self.engine_name == "6s":
errors.append("6s requires earth_sun_distance_file to be specified")
if self.irradiance_file is None and self.engine_name == "6s":
errors.append("6s requires irradiance_file to be specified")
if self.engine_name == "sRTMnet":
if self.emulator_file is None:
# Fallback to the path specified by the isofit.ini
self.emulator_file = env.path("srtmnet", env["srtmnet.file"])
if not self.emulator_file.exists():
errors.append(
"The sRTMnet requires an emulator_file to be specified."
)
if (
(os.path.splitext(self.emulator_file)[1] != ".h5")
and (os.path.splitext(self.emulator_file)[1] != ".npz")
and (os.path.splitext(self.emulator_file)[1] != ".6c")
):
errors.append(
"sRTMnet now requires the emulator_file to be of type .h5 (or .npz for experimental 6c emulator). "
"Please download an updated version from:\n https://zenodo.org/records/10831425"
)
if self.emulator_file.endswith(".6c"):
from isofit.radiative_transfer.engines.six_s import get_exe
if "co2" not in get_exe(self.engine_base_dir, version=True):
errors.append(
"sRTMnet 6C requires a CO2 version of 6S. Please use the isofit download CLI to pull a CO2 tag: https://github.com/isofit/6S/tags"
)
if self.emulator_aux_file is None:
# Fallback to the path specified by the isofit.ini
self.emulator_aux_file = env.path("srtmnet", env["srtmnet.aux"])
if not self.emulator_aux_file.exists():
errors.append(
"The sRTMnet requires an emulator_aux_file to be specified."
)
files = [
self.obs_file,
self.aerosol_model_file,
self.aerosol_template_file,
]
for file in files:
if file is not None and not os.path.isfile(file):
errors.append(
f"Radiative transfer engine file not found on system: {file}"
)
if isinstance(self.lut_complevel, int) and self.lut_complevel < 1:
errors.append("The LUT complevel must be and int greater than 0")
return errors
[docs]
class RadiativeTransferUnknownsConfig(BaseConfigSection):
"""
Radiative transfer unknowns configuration.
"""
def __init__(self, sub_configdic: dict = None):
super().__init__()
[docs]
self._H2O_ABSCO_type = float
self.set_config_options(sub_configdic)
[docs]
def _check_config_validity(self) -> List[str]:
errors = list()
return errors
[docs]
class RadiativeTransferConfig(BaseConfigSection):
"""
Forward model configuration.
"""
def __init__(self, sub_configdic: dict = None):
[docs]
self._statevector_type = StateVectorConfig
[docs]
self.statevector: StateVectorConfig = StateVectorConfig({})
[docs]
self._lut_grid_type = OrderedDict
[docs]
self._unknowns_type = RadiativeTransferUnknownsConfig
[docs]
self.unknowns: RadiativeTransferUnknownsConfig = None
[docs]
self._interpolator_style_type = str
[docs]
self.interpolator_style = "mlg"
"""str: Style of interpolation.
- mlg = Multilinear Grid
- rg = RegularGrid
Speed performance:
mlg >> stacked rg >> unstacked rg
Caching provides significant gains for rg, marginal for mlg"""
[docs]
self._overwrite_interpolator_type = bool
[docs]
self.overwrite_interpolator = False
"""bool: Overwrite any existing interpolator pickles"""
[docs]
self._cache_size_type = int
"""int: Size of the cache to store interpolation lookups. Defaults to 16 which
provides the most significant gains. Setting higher may provide marginal gains."""
self.set_config_options(sub_configdic)
# sort lut_grid
for key, value in self.lut_grid.items():
self.lut_grid[key] = sorted(self.lut_grid[key])
self.lut_grid = OrderedDict(sorted(self.lut_grid.items(), key=lambda t: t[0]))
# Hold this parameter for after the config_options, as radiative_transfer_engines
# have a special (dynamic) load
[docs]
self._radiative_transfer_engines_type = list()
[docs]
self.radiative_transfer_engines = []
self._set_rt_config_options(sub_configdic["radiative_transfer_engines"])
[docs]
def _set_rt_config_options(self, subconfig):
if type(subconfig) is list:
for rte in subconfig:
rt_model = RadiativeTransferEngineConfig(rte)
self.radiative_transfer_engines.append(rt_model)
elif type(subconfig) is dict:
for key in subconfig:
rt_model = RadiativeTransferEngineConfig(subconfig[key], name=key)
self.radiative_transfer_engines.append(rt_model)
[docs]
def _check_config_validity(self) -> List[str]:
errors = list()
for key, item in self.lut_grid.items():
if len(item) < 2:
errors.append(
"lut_grid item {} has less than the required 2 elements".format(key)
)
if np.unique(item).size < len(item):
errors.append(f"Detected duplicate values in lut_grid item {key}")
for rte in self.radiative_transfer_engines:
errors.extend(rte.check_config_validity())
kinds = ["rg", "nds", "mlg"] # Implemented kinds of interpolator functions
kind = self.interpolator_style[:3]
degrees = self.interpolator_style[4:]
if kind not in kinds:
errors.append(
f"Interpolator style {self.interpolator_style} must be one of: {kinds}"
)
if kind == "nds":
try:
degree = int(degrees)
assert degree >= 0 and np.isfinite(degree)
except:
errors.append(
"Invalid degree number. Should be an integer, e.g. nds-3, got"
f" {degrees!r} from {self.interpolator_style!r}[4:]"
)
return errors