Commit 8b751747 authored by Cresson Remi's avatar Cresson Remi
Browse files

REFAC: remove DRS namespace + create aliases in __init__

1 merge request!16REFAC: remove DRS namespace + create aliases in __init__
Pipeline #34246 failed with stages
in 58 seconds
Showing with 101 additions and 95 deletions
+101 -95
# -*- coding: utf-8 -*-
"""
This module helps to process local remote sensing products
"""
__version__ = "1.0.1"
from .core import load_scenes, save_scenes
from .indexation import Index
from .download import TheiaDownloader
\ No newline at end of file
""" """
This module provides the Source class, which aims to handle the access to the rasters delivered by EO products This module provides the Source class, which aims to handle the access to the rasters delivered by EO products
""" """
import pickle
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
import datetime import datetime
import pyotb import pyotb
from scenes import utils from scenes import utils
def save_scenes(scenes_list, pickle_file):
"""
Use pickle to save scenes
:param scenes_list: a list of Scene instances
:param pickle_file: pickle file
"""
pickle.dump(scenes_list, open(pickle_file, "wb"))
def load_scenes(pickle_file):
"""
Use pickle to save Spot-6/7 scenes
:param pickle_file: pickle file
:return: list of Scene instances
"""
return pickle.load(open(pickle_file, "rb"))
class Source(pyotb.Output): class Source(pyotb.Output):
""" """
Source class. Source class.
......
"""
A set of utils to deal with DRS products
"""
import pickle
import tqdm
from scenes import spot, utils
def find_all_dimaps(pth):
"""
Return the list of DIMAPS that are inside all subdirectories of the root directory
:param pth: root directory
:return: list of DIMAPS
"""
return utils.find_files_in_all_subdirs(pth=pth, pattern="DIM_*.XML")
def get_spot67_scenes(root_dir):
"""
Return the list of pairs of PAN/XS DIMAPS
:param root_dir: directory containing "MS" and "PAN" subdirectories
:return: list of Spot67Scenes instances
"""
# List files
look_dir = root_dir + "/MS"
print("Find files in {}".format(look_dir))
dimap_xs_files = find_all_dimaps(look_dir)
print("Found {} DIMAP files in MS folder".format(len(dimap_xs_files)), flush=True)
# Create scenes list
scenes = []
errors = {}
for dimap_file_xs in tqdm.tqdm(dimap_xs_files):
try:
# Find pairs of XS/PAN DIMAPS
pan_path = dimap_file_xs[:dimap_file_xs.find("/PROD_SPOT")]
pan_path = pan_path.replace("/MS/", "/PAN/")
pan_path = pan_path.replace("_MS_", "_PAN_")
dimap_pan_files = find_all_dimaps(pan_path)
nb_files = len(dimap_pan_files)
if nb_files != 1:
raise ValueError("{} DIMAPS candidates found in {} ".format(nb_files, pan_path))
dimap_file_pan = dimap_pan_files[0]
# Instantiate a new scene object
new_scene = spot.Spot67Scene(dimap_file_pan=dimap_file_pan, dimap_file_xs=dimap_file_xs)
scenes.append(new_scene)
except Exception as error:
if dimap_file_xs not in errors:
errors[dimap_file_xs] = []
errors[dimap_file_xs].append(error)
raise
print("{} scenes imported".format(len(scenes)))
if errors:
print("{} scenes could not have been imported.".format(len(errors)))
for dimap_file_xs, error_list in errors.items():
print("Errors for {}:".format(dimap_file_xs))
for error in error_list:
print("\t{}".format(error))
return scenes
def save_scenes(scenes_list, pickle_file):
"""
Use pickle to save scenes
:param scenes_list: a list of Scene instances
:param pickle_file: pickle file
"""
pickle.dump(scenes_list, open(pickle_file, "wb"))
def load_scenes(pickle_file):
"""
Use pickle to save Spot-6/7 scenes
:param pickle_file: pickle file
:return: list of Scene instances
"""
return pickle.load(open(pickle_file, "rb"))
...@@ -2,12 +2,69 @@ ...@@ -2,12 +2,69 @@
Spot 6/7 root_scene class Spot 6/7 root_scene class
""" """
import datetime import datetime
import tqdm
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
import pyotb import pyotb
from scenes import utils from scenes import utils
from scenes.core import Source, Imagery, Scene from scenes.core import Source, Imagery, Scene
def find_all_dimaps(pth):
"""
Return the list of DIMAPS that are inside all subdirectories of the root directory
:param pth: root directory
:return: list of DIMAPS
"""
return utils.find_files_in_all_subdirs(pth=pth, pattern="DIM_*.XML")
def get_spot67_scenes(root_dir):
"""
Return the list of pairs of PAN/XS DIMAPS
:param root_dir: directory containing "MS" and "PAN" subdirectories
:return: list of Spot67Scenes instances
"""
# List files
look_dir = root_dir + "/MS"
print("Find files in {}".format(look_dir))
dimap_xs_files = find_all_dimaps(look_dir)
print("Found {} DIMAP files in MS folder".format(len(dimap_xs_files)), flush=True)
# Create scenes list
scenes = []
errors = {}
for dimap_file_xs in tqdm.tqdm(dimap_xs_files):
try:
# Find pairs of XS/PAN DIMAPS
pan_path = dimap_file_xs[:dimap_file_xs.find("/PROD_SPOT")]
pan_path = pan_path.replace("/MS/", "/PAN/")
pan_path = pan_path.replace("_MS_", "_PAN_")
dimap_pan_files = find_all_dimaps(pan_path)
nb_files = len(dimap_pan_files)
if nb_files != 1:
raise ValueError("{} DIMAPS candidates found in {} ".format(nb_files, pan_path))
dimap_file_pan = dimap_pan_files[0]
# Instantiate a new scene object
new_scene = spot.Spot67Scene(dimap_file_pan=dimap_file_pan, dimap_file_xs=dimap_file_xs)
scenes.append(new_scene)
except Exception as error:
if dimap_file_xs not in errors:
errors[dimap_file_xs] = []
errors[dimap_file_xs].append(error)
raise
print("{} scenes imported".format(len(scenes)))
if errors:
print("{} scenes could not have been imported.".format(len(errors)))
for dimap_file_xs, error_list in errors.items():
print("Errors for {}:".format(dimap_file_xs))
for error in error_list:
print("\t{}".format(error))
return scenes
class Spot67Source(Source): class Spot67Source(Source):
""" """
Spot 6/7 source class Spot 6/7 source class
......
import argparse import argparse
from scenes import download, utils from scenes import TheiaDownloader, utils
import datetime import datetime
# Arguments # Arguments
...@@ -15,5 +15,5 @@ params = parser.parse_args() ...@@ -15,5 +15,5 @@ params = parser.parse_args()
# Get all scenes in the root_dir # Get all scenes in the root_dir
_, _, bbox = utils.get_epsg_extent_bbox(params.refimage) _, _, bbox = utils.get_epsg_extent_bbox(params.refimage)
acq_date = datetime.datetime(year=params.year, month=params.month, day=params.day) acq_date = datetime.datetime(year=params.year, month=params.month, day=params.day)
downloader = download.TheiaDownloader(config_file=params.theia_cfg) downloader = TheiaDownloader(config_file=params.theia_cfg)
downloader.download_closest(bbox_wgs84=bbox, acq_date=acq_date, download_dir=params.download_dir) downloader.download_closest(bbox_wgs84=bbox, acq_date=acq_date, download_dir=params.download_dir)
import argparse import argparse
from scenes import drs from scenes import save_scenes
from scenes.spot import get_spot67_scenes
# Arguments # Arguments
parser = argparse.ArgumentParser(description="Test",) parser = argparse.ArgumentParser(description="Test",)
...@@ -10,7 +11,7 @@ params = parser.parse_args() ...@@ -10,7 +11,7 @@ params = parser.parse_args()
# Get all scenes in the root_dir # Get all scenes in the root_dir
scenes = [] scenes = []
for root_dir in params.root_dirs: for root_dir in params.root_dirs:
scenes += drs.get_spot67_scenes(root_dir) scenes += get_spot67_scenes(root_dir)
# Save scenes in a pickle file # Save scenes in a pickle file
drs.save_scenes(scenes, params.out_pickle) save_scenes(scenes, params.out_pickle)
import argparse import argparse
import scenes.indexation as indexation
import gdal import gdal
from scenes import utils, drs from scenes import utils, Index
# Arguments # Arguments
parser = argparse.ArgumentParser(description="Test",) parser = argparse.ArgumentParser(description="Test",)
...@@ -14,7 +13,7 @@ scenes = drs.get_spot67_scenes(params.root_dir) ...@@ -14,7 +13,7 @@ scenes = drs.get_spot67_scenes(params.root_dir)
# spatial index # spatial index
print("Indexation...") print("Indexation...")
idx = indexation.Index(scenes) idx = Index(scenes)
# search # search
print("search roi") print("search roi")
......
import argparse import argparse
import gdal import gdal
import pyotb import pyotb
from scenes import utils, drs, indexation from scenes import utils, Index
from scenes.spot import get_spot67_scenes
# Arguments # Arguments
parser = argparse.ArgumentParser(description="Test",) parser = argparse.ArgumentParser(description="Test",)
...@@ -11,11 +12,11 @@ parser.add_argument("--out", required=True) ...@@ -11,11 +12,11 @@ parser.add_argument("--out", required=True)
params = parser.parse_args() params = parser.parse_args()
# Find pairs of DIMAPS # Find pairs of DIMAPS
scenes = drs.get_spot67_scenes(params.root_dir) scenes = get_spot67_scenes(params.root_dir)
# spatial index # spatial index
print("Indexation...") print("Indexation...")
idx = indexation.Index(scenes) idx = Index(scenes)
# search # search
print("search roi") print("search roi")
......
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
from scenes_test_base import ScenesTestBase from scenes_test_base import ScenesTestBase
from scenes import indexation, utils from scenes import Index, utils
import tests_data import tests_data
class ImageryTest(ScenesTestBase): class ImageryTest(ScenesTestBase):
def test_scene1_indexation(self): def test_scene1_indexation(self):
index = indexation.Index(scenes_list=[tests_data.SCENE1]) index = Index(scenes_list=[tests_data.SCENE1])
self.assertTrue(index.find(bbox_wgs84=(43.706, 43.708, 4.317, 4.420))) self.assertTrue(index.find(bbox_wgs84=(43.706, 43.708, 4.317, 4.420)))
self.assertFalse(index.find(bbox_wgs84=(43.000, 43.001, 3.000, 3.001))) self.assertFalse(index.find(bbox_wgs84=(43.000, 43.001, 3.000, 3.001)))
self.assertTrue(index.find(bbox_wgs84=utils.get_bbox_wgs84_from_vector(tests_data.ROI_MTP_4326))) self.assertTrue(index.find(bbox_wgs84=utils.get_bbox_wgs84_from_vector(tests_data.ROI_MTP_4326)))
......
import argparse import argparse
from scenes import utils, sentinel, drs from scenes import utils, sentinel, save_scenes
# Arguments # Arguments
parser = argparse.ArgumentParser(description="Test",) parser = argparse.ArgumentParser(description="Test",)
...@@ -16,4 +16,4 @@ for root_dir in params.root_dirs: ...@@ -16,4 +16,4 @@ for root_dir in params.root_dirs:
scenes = [product_type(archive=archive) for archive in archives] scenes = [product_type(archive=archive) for archive in archives]
# Save scenes in a pickle file # Save scenes in a pickle file
drs.save_scenes(scenes, params.out_pickle) save_scenes(scenes, params.out_pickle)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment