"""
Grab specific sentinel2 images for testing
"""
import ubelt as ub
import os
import pathlib
from glob import glob
[docs]
def bands_sentinel2(entry):
"""
Get only the band files from a Sentinel-2 RasterMetaEntry downloaded from RGD.
Args:
entry: RasterMetaEntry, the output type of rgdc.download_raster
Returns:
list of pathlib paths to band files
"""
return [str(p) for p in entry.images if p.match('*_B*.jp2')]
[docs]
def grab_sentinel2_product(index=0, overwrite=False):
"""
Download and cache all items for a Sentinel-2 product.
TODO when RGD supports API keys, give one to this repo and use that instead of fels.
Args:
index: 0, 1, or 2. Currently this function just picks 3 scenes over KR in Nov 2018.
overwrite (bool, default=False): if True, always downloads the files
Returns:
rgd_client.rgdc.RasterDownload(
path: pathlib.Path,
images: List[pathlib.Path],
ancillary: List[pathlib.Path]
)
Example:
>>> # xdoctest: +SKIP("too many https errors")
>>> # xdoctest: +REQUIRES(--network)
>>> from geowatch.demo.sentinel2_demodata import * # NOQA
>>> product = grab_sentinel2_product()
>>> assert len(bands_sentinel2(product)) == 13
"""
try:
from rgd_imagery_client import RasterDownload
except ImportError:
from rgd_client.rgdc import RasterDownload
from os.path import join, basename, relpath
urls = [
'http://storage.googleapis.com/gcp-public-data-sentinel-2/tiles/52/S/DG/S2A_MSIL1C_20181101T020821_N0206_R103_T52SDG_20181101T040328.SAFE',
'http://storage.googleapis.com/gcp-public-data-sentinel-2/tiles/52/S/DG/S2A_MSIL1C_20181104T021841_N0206_R003_T52SDG_20181104T055443.SAFE',
'http://storage.googleapis.com/gcp-public-data-sentinel-2/tiles/52/S/DG/S2B_MSIL1C_20181106T020849_N0207_R103_T52SDG_20181106T034331.SAFE'
]
url = urls[index]
# By default cache to the $XDG_CACHE_HOME/geowatch
dset_dpath = ub.Path.appdir('geowatch/demo/grab_s2').ensuredir()
# Cache the scene using the same path used by google cloud storage
tile_hierarchy = os.path.sep.join(url.split('tiles/')[1].split('/')[:3])
scene_dpath = ub.ensuredir((dset_dpath, tile_hierarchy))
safe_dpath = pathlib.Path(scene_dpath) / basename(url)
was_failed_download = (
not safe_dpath.exists() or not list(safe_dpath.glob('*'))
)
"""
# We could vendor code from fels if we wanted.
import liberator
from fels.sentinel2 import get_sentinel2_image
lib = liberator.Liberator()
lib.add_dynamic(get_sentinel2_image)
print(lib.current_sourcecode())
"""
try:
from fels.sentinel2 import get_sentinel2_image
except ImportError:
from fels import get_sentinel2_image
if was_failed_download:
# This is really slow even if the data is cached with default fels
# This PR: https://github.com/vascobnunes/fetchLandsatSentinelFromGoogleCloud/pull/58
# can help speed it up, which is merged but not releases as of
# 2021-10-30
# Download the scene
assert get_sentinel2_image(url,
scene_dpath,
# overwrite=was_failed_download,
overwrite=True, # makes this really slow
reject_old=True)
# Build a rgdc object to return the scene
root = join(scene_dpath, url.split('/')[-1])
fpaths = sorted(glob(join(root, '**', '*.*'), recursive=True))
fpaths = [pathlib.Path(relpath(f, start=root)) for f in fpaths]
product = RasterDownload(
path=pathlib.Path(root),
images=[f for f in fpaths if f.suffix == '.jp2'],
ancillary=[f for f in fpaths if f.suffix != '.jp2'])
return product