Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Sanitize metadata from the file to use SIFT's Enums
if "name" in gtiff_meta:
gtiff_meta[Info.DATASET_NAME] = gtiff_meta.pop("name")
if "platform" in gtiff_meta:
plat = gtiff_meta.pop("platform")
try:
gtiff_meta[Info.PLATFORM] = Platform(plat)
except ValueError:
gtiff_meta[Info.PLATFORM] = Platform.UNKNOWN
LOG.warning("Unknown platform being loaded: {}".format(plat))
if "instrument" in gtiff_meta or "sensor" in gtiff_meta:
inst = gtiff_meta.pop("sensor", gtiff_meta.pop("instrument", None))
try:
gtiff_meta[Info.INSTRUMENT] = Instrument(inst)
except ValueError:
gtiff_meta[Info.INSTRUMENT] = Instrument.UNKNOWN
LOG.warning("Unknown instrument being loaded: {}".format(inst))
if "start_time" in gtiff_meta:
start_time = datetime.strptime(gtiff_meta["start_time"], "%Y-%m-%dT%H:%M:%SZ")
gtiff_meta[Info.SCHED_TIME] = start_time
gtiff_meta[Info.OBS_TIME] = start_time
if "end_time" in gtiff_meta:
end_time = datetime.strptime(gtiff_meta["end_time"], "%Y-%m-%dT%H:%M:%SZ")
gtiff_meta[Info.OBS_DURATION] = end_time - start_time
if "valid_min" in gtiff_meta:
gtiff_meta["valid_min"] = float(gtiff_meta["valid_min"])
if "valid_max" in gtiff_meta:
gtiff_meta["valid_max"] = float(gtiff_meta["valid_max"])
if "standard_name" in gtiff_meta:
gtiff_meta[Info.STANDARD_NAME] = gtiff_meta["standard_name"]
if "flag_values" in gtiff_meta:
gtiff_meta["flag_values"] = tuple(int(x) for x in gtiff_meta["flag_values"].split(','))
def _get_platform_instrument(attrs: dict):
"""Convert SatPy platform_name/sensor to """
attrs[Info.INSTRUMENT] = attrs.get('sensor')
attrs[Info.PLATFORM] = attrs.get('platform_name')
# Special handling of GRIB forecast data
if 'centreDescription' in attrs and \
attrs[Info.INSTRUMENT] == 'unknown':
description = attrs['centreDescription']
if attrs.get(Info.PLATFORM) is None:
attrs[Info.PLATFORM] = 'NWP'
if 'NCEP' in description:
attrs[Info.INSTRUMENT] = 'GFS'
if attrs[Info.INSTRUMENT] in ['GFS', 'unknown']:
attrs[Info.INSTRUMENT] = Instrument.GFS
if attrs[Info.PLATFORM] in ['NWP', 'unknown']:
attrs[Info.PLATFORM] = Platform.NWP
# FUTURE: Use standard string names for platform/instrument
# instead of an Enum. Otherwise, could use a reverse
# Enum lookup to match Enum values to Enum keys.
# if we haven't figured out what these are then give up and say they are unknown
if isinstance(attrs[Info.PLATFORM], str):
plat_str = attrs[Info.PLATFORM].lower().replace('-', '')
attrs[Info.PLATFORM] = PLATFORM_MAP.get(plat_str, attrs[Info.PLATFORM])
if not attrs[Info.PLATFORM] or isinstance(attrs[Info.PLATFORM], str):
attrs[Info.PLATFORM] = Platform.UNKNOWN
if isinstance(attrs[Info.INSTRUMENT], str):
inst_str = attrs[Info.INSTRUMENT].lower().replace('-', '')
attrs[Info.INSTRUMENT] = INSTRUMENT_MAP.get(inst_str, attrs[Info.INSTRUMENT])
def _check_geotiff_metadata(gtiff):
gtiff_meta = gtiff.GetMetadata()
# Sanitize metadata from the file to use SIFT's Enums
if "name" in gtiff_meta:
gtiff_meta[Info.DATASET_NAME] = gtiff_meta.pop("name")
if "platform" in gtiff_meta:
plat = gtiff_meta.pop("platform")
try:
gtiff_meta[Info.PLATFORM] = Platform(plat)
except ValueError:
gtiff_meta[Info.PLATFORM] = Platform.UNKNOWN
LOG.warning("Unknown platform being loaded: {}".format(plat))
if "instrument" in gtiff_meta or "sensor" in gtiff_meta:
inst = gtiff_meta.pop("sensor", gtiff_meta.pop("instrument", None))
try:
gtiff_meta[Info.INSTRUMENT] = Instrument(inst)
except ValueError:
gtiff_meta[Info.INSTRUMENT] = Instrument.UNKNOWN
LOG.warning("Unknown instrument being loaded: {}".format(inst))
if "start_time" in gtiff_meta:
start_time = datetime.strptime(gtiff_meta["start_time"], "%Y-%m-%dT%H:%M:%SZ")
gtiff_meta[Info.SCHED_TIME] = start_time
gtiff_meta[Info.OBS_TIME] = start_time
if "end_time" in gtiff_meta:
end_time = datetime.strptime(gtiff_meta["end_time"], "%Y-%m-%dT%H:%M:%SZ")
gtiff_meta[Info.OBS_DURATION] = end_time - start_time
if "valid_min" in gtiff_meta:
gtiff_meta["valid_min"] = float(gtiff_meta["valid_min"])
if "valid_max" in gtiff_meta:
gtiff_meta["valid_max"] = float(gtiff_meta["valid_max"])
if "standard_name" in gtiff_meta:
gtiff_meta[Info.STANDARD_NAME] = gtiff_meta["standard_name"]
"""
return None, None
DEFAULT_COLORMAPS = {
'toa_bidirectional_reflectance': DEFAULT_VIS,
'toa_brightness_temperature': DEFAULT_IR,
'brightness_temperature': DEFAULT_IR,
'height_at_cloud_top': 'Cloud Top Height',
'air_temperature': DEFAULT_IR,
'relative_humidity': DEFAULT_IR,
# 'thermodynamic_phase_of_cloud_water_particles_at_cloud_top': 'Cloud Phase',
}
_NW_GOESR_ABI = {
Instrument.ABI: { # http://www.goes-r.gov/education/ABI-bands-quick-info.html
1: 0.47,
2: 0.64,
3: 0.86,
4: 1.37,
5: 1.6,
6: 2.2,
7: 3.9,
8: 6.2,
9: 6.9,
10: 7.3,
11: 8.4,
12: 9.6,
13: 10.3,
14: 11.2,
15: 12.3,
16: 13.3,
plat, yyyymmdd, hhmm, bb, scene = m.groups()
when = datetime.strptime(yyyymmdd + hhmm, '%Y%m%d%H%M')
plat = Platform('Himawari-{}'.format(int(plat)))
band = int(bb)
#
# # workaround to make old files work with new information
# from uwsift.model.guidebook import AHI_HSF_Guidebook
# if band in AHI_HSF_Guidebook.REFL_BANDS:
# standard_name = "toa_bidirectional_reflectance"
# else:
# standard_name = "toa_brightness_temperature"
meta.update({
Info.PLATFORM: plat,
Info.BAND: band,
Info.INSTRUMENT: Instrument.AHI,
Info.SCHED_TIME: when,
Info.OBS_TIME: when,
Info.OBS_DURATION: DEFAULT_GTIFF_OBS_DURATION,
Info.SCENE: scene,
})
return meta
ARITHMETIC = 2
class Instrument(Enum):
UNKNOWN = '???'
AHI = 'AHI'
ABI = 'ABI'
AMI = 'AMI'
GFS = 'GFS'
NAM = 'NAM'
SEVIRI = 'SEVIRI'
LI = 'LI'
GLM = 'GLM'
INSTRUMENT_MAP = {v.value.lower().replace('-', ''): v for v in Instrument}
class Platform(Enum):
UNKNOWN = '???'
HIMAWARI_8 = 'Himawari-8'
HIMAWARI_9 = 'Himawari-9'
GOES_16 = 'G16'
GOES_17 = 'G17'
GOES_18 = 'G18'
NWP = 'NWP'
MSG8 = 'Meteosat-8'
MSG9 = 'Meteosat-9'
MSG10 = 'Meteosat-10'
MSG11 = 'Meteosat-11'
GK2A = "GEO-KOMPSAT-2A"
def _basic_pug_metadata(pug):
return {
Info.PLATFORM: PLATFORM_ID_TO_PLATFORM[pug.platform_id], # e.g. G16, H8
Info.BAND: pug.band,
Info.DATASET_NAME: 'B{:02d}'.format(pug.band),
Info.INSTRUMENT: Instrument.AHI if 'Himawari' in pug.instrument_type else Instrument.ABI,
Info.SCHED_TIME: pug.sched_time,
Info.OBS_TIME: pug.time_span[0],
Info.OBS_DURATION: pug.time_span[1] - pug.time_span[0],
Info.DISPLAY_TIME: pug.display_time,
Info.SCENE: pug.scene_id,
Info.DISPLAY_NAME: pug.display_name,
}
}
# Instrument -> Band Number -> Nominal Wavelength
NOMINAL_WAVELENGTHS = {
Platform.HIMAWARI_8: _NW_HIMAWARI_AHI,
Platform.HIMAWARI_9: _NW_HIMAWARI_AHI,
Platform.GOES_16: _NW_GOESR_ABI,
Platform.GOES_17: _NW_GOESR_ABI,
}
# CF compliant Standard Names (should be provided by input files or the workspace)
# Instrument -> Band Number -> Standard Name
_SN_GOESR_ABI = {
Instrument.ABI: {
1: "toa_bidirectional_reflectance",
2: "toa_bidirectional_reflectance",
3: "toa_bidirectional_reflectance",
4: "toa_bidirectional_reflectance",
5: "toa_bidirectional_reflectance",
6: "toa_bidirectional_reflectance",
7: "toa_brightness_temperature",
8: "toa_brightness_temperature",
9: "toa_brightness_temperature",
10: "toa_brightness_temperature",
11: "toa_brightness_temperature",
12: "toa_brightness_temperature",
13: "toa_brightness_temperature",
14: "toa_brightness_temperature",
15: "toa_brightness_temperature",
16: "toa_brightness_temperature",