Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _metadata_for_path(pathname):
meta = {}
if not pathname:
return meta
# Old but still necesary, get some information from the filename instead of the content
m = re.match(r'HS_H(\d\d)_(\d{8})_(\d{4})_B(\d\d)_([A-Za-z0-9]+).*', os.path.split(pathname)[1])
if m is not None:
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,
data=img_data)
yield zult
# Finally, update content mtime and atime
c.atime = c.mtime = datetime.utcnow()
# self._S.commit()
# map .platform_id in PUG format files to SIFT platform enum
PLATFORM_ID_TO_PLATFORM = {
'G16': Platform.GOES_16,
'G17': Platform.GOES_17,
# hsd2nc export of AHI data as PUG format
'Himawari-8': Platform.HIMAWARI_8,
'Himawari-9': Platform.HIMAWARI_9,
# axi2cmi export as PUG, more consistent with other uses
'H8': Platform.HIMAWARI_8,
'H9': Platform.HIMAWARI_9
}
class GoesRPUGImporter(aSingleFileWithSingleProductImporter):
"""
Import from PUG format GOES-16 netCDF4 files
"""
@staticmethod
def _basic_pug_metadata(pug):
return {
Info.PLATFORM: PLATFORM_ID_TO_PLATFORM[pug.platform_id], # e.g. G16, H8
Info.BAND: pug.band,
from satpy.dataset import DatasetID
except ImportError:
LOG.warning("SatPy is not installed and will not be used for importing.")
Scene = None
DatasetID = None
try:
from skimage.measure import find_contours
except ImportError:
find_contours = None
DEFAULT_GTIFF_OBS_DURATION = timedelta(seconds=60)
DEFAULT_GUIDEBOOK = ABI_AHI_Guidebook
GUIDEBOOKS = {
Platform.GOES_16: ABI_AHI_Guidebook,
Platform.GOES_17: ABI_AHI_Guidebook,
Platform.HIMAWARI_8: ABI_AHI_Guidebook,
Platform.HIMAWARI_9: ABI_AHI_Guidebook,
}
import_progress = namedtuple('import_progress',
['uuid', 'stages', 'current_stage', 'completion', 'stage_desc', 'dataset_info', 'data'])
"""
# stages:int, number of stages this import requires
# current_stage:int, 0..stages-1 , which stage we're on
# completion:float, 0..1 how far we are along on this stage
# stage_desc:tuple(str), brief description of each of the stages we'll be doing
"""
def _load_satpy_readers_cache(force_refresh=None):
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")
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"
PLATFORM_MAP = {v.value.lower().replace('-', ''): v for v in Platform}
PLATFORM_MAP['h8'] = Platform.HIMAWARI_8
PLATFORM_MAP['h9'] = Platform.HIMAWARI_9
PLATFORM_MAP['goes16'] = Platform.GOES_16
PLATFORM_MAP['goes17'] = Platform.GOES_17
PLATFORM_MAP['goes18'] = Platform.GOES_18
class Info(Enum):
"""
Standard keys for info dictionaries
Note: some fields correspond to database fields in workspace.metadatabase !
"""
UNKNOWN = '???'
# full path to the resource that the file came from
# DEPRECATED since datasets may not have one-to-one pathname mapping
PATHNAME = 'path'
# CF content
SHORT_NAME = 'short_name' # CF short_name
LONG_NAME = 'long_name' # CF long_name
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:
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",
},
}
STANDARD_NAMES = {
Platform.HIMAWARI_8: _SN_HIMAWARI_AHI,
Platform.HIMAWARI_9: _SN_HIMAWARI_AHI,
Platform.GOES_16: _SN_GOESR_ABI,
Platform.GOES_17: _SN_GOESR_ABI,
}
BT_STANDARD_NAMES = ["toa_brightness_temperature", 'brightness_temperature', 'air_temperature']
class ABI_AHI_Guidebook(Guidebook):
"e.g. HS_H08_20150714_0030_B10_FLDK_R20.merc.tif"
_cache = None # {uuid:metadata-dictionary, ...}
def __init__(self):
self._cache = {}
def collect_info(self, info):
"""Collect information that may not come from the dataset.
This method should only be called once to "fill in" metadata
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"
PLATFORM_MAP = {v.value.lower().replace('-', ''): v for v in Platform}
PLATFORM_MAP['h8'] = Platform.HIMAWARI_8
PLATFORM_MAP['h9'] = Platform.HIMAWARI_9
PLATFORM_MAP['goes16'] = Platform.GOES_16
PLATFORM_MAP['goes17'] = Platform.GOES_17
PLATFORM_MAP['goes18'] = Platform.GOES_18
class Info(Enum):
"""
Standard keys for info dictionaries
Note: some fields correspond to database fields in workspace.metadatabase !
"""
UNKNOWN = '???'
# full path to the resource that the file came from
# DEPRECATED since datasets may not have one-to-one pathname mapping
PATHNAME = 'path'