diff --git a/pyotb/core.py b/pyotb/core.py index 71a032c4095819d6125742f7cd6ad65c3881a4f2..0047dd5566e19616568ccaccc2e06592ca4fdbcb 100644 --- a/pyotb/core.py +++ b/pyotb/core.py @@ -2,6 +2,7 @@ """This module is the core of pyotb.""" from __future__ import annotations +import re from abc import ABC, abstractmethod from ast import literal_eval from pathlib import Path @@ -731,10 +732,7 @@ class App(OTBObject): ) try: if self.is_input(key): - if self.is_key_images_list(key): - self.__set_param(key, [add_vsi_prefix(p) for p in obj]) - else: - self.__set_param(key, add_vsi_prefix(obj)) + self.__set_param(key, self.__check_input_param(obj)) else: self.__set_param(key, obj) except (RuntimeError, TypeError, ValueError, KeyError) as e: @@ -969,6 +967,49 @@ class App(OTBObject): kwargs.update({self.input_key: arg}) return kwargs + def __check_input_param( + self, obj: list | OTBObject | str | Path + ) -> list | OTBObject | str: + """Check the type and value of an input param. + + Args: + obj: input parameter value + + Returns: + object, string with new /vsi prefix(es) if needed + + """ + if isinstance(obj, list): + return [self.__check_input_param(o) for o in obj] + # May be we could add some checks here + if isinstance(obj, OTBObject): + return obj + if isinstance(obj, Path): + obj = str(obj) + if isinstance(obj, str): + if not obj.startswith("/vsi"): + # Remote file. TODO: add support for S3 / GS / AZ + if obj.startswith(("https://", "http://", "ftp://")): + obj = "/vsicurl/" + obj + # Compressed file + prefixes = { + ".tar": "vsitar", + ".tar.gz": "vsitar", + ".tgz": "vsitar", + ".gz": "vsigzip", + ".7z": "vsi7z", + ".zip": "vsizip", + ".rar": "vsirar", + } + expr = r"(.*?)(\.7z|\.zip|\.rar|\.tar\.gz|\.tgz|\.tar|\.gz)(.*)" + parts = re.match(expr, obj) + if parts: + file, ext = parts.group(1), parts.group(2) + if not Path(file + ext).is_dir(): + obj = f"/{prefixes[ext]}/{obj}" + return obj + raise TypeError(f"{self.name}: wrong input parameter type ({type(obj)})") + def __set_param( self, key: str, obj: list | tuple | OTBObject | otb.Application | list[Any] ): @@ -1580,38 +1621,6 @@ class Output(OTBObject): return str(self.filepath) -def add_vsi_prefix(filepath: str | Path) -> str: - """Append vsi prefixes to file URL or path if needed. - - Args: - filepath: file path or URL - - Returns: - string with new /vsi prefix(es) - - """ - if isinstance(filepath, Path): - filepath = str(filepath) - if isinstance(filepath, str) and not filepath.startswith("/vsi"): - # Remote file. TODO: add support for S3 / GS / AZ - if filepath.startswith(("https://", "http://", "ftp://")): - filepath = "/vsicurl/" + filepath - # Compressed file - prefixes = { - ".tar": "vsitar", - ".tgz": "vsitar", - ".gz": "vsigzip", - ".7z": "vsi7z", - ".zip": "vsizip", - ".rar": "vsirar", - } - basename = filepath.split("?")[0] - ext = Path(basename).suffix - if ext in prefixes: - filepath = f"/{prefixes[ext]}/{filepath}" - return filepath - - def get_nbchannels(inp: str | Path | OTBObject) -> int: """Get the nb of bands of input image. diff --git a/tests/test_core.py b/tests/test_core.py index 82173cda265c13f6686f1556e41235bdd264370f..3bfc5dd7704099f154745fbb3b2d6ec34f5f1883 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -49,17 +49,23 @@ def test_app_properties(): def test_app_input_vsi(): + # Ensure old way is still working: ExtractROI will raise RuntimeError if a path is malformed + pyotb.Input("/vsicurl/" + SPOT_IMG_URL) # Simple remote file info = pyotb.ReadImageInfo("https://fake.com/image.tif", frozen=True) assert info.app.GetParameterValue("in") == "/vsicurl/https://fake.com/image.tif" assert info.parameters["in"] == "https://fake.com/image.tif" - # Compressed remote file - info = pyotb.ReadImageInfo("https://fake.com/image.tif.zip", frozen=True) + # Compressed single file archive + info = pyotb.ReadImageInfo("image.tif.zip", frozen=True) + assert info.app.GetParameterValue("in") == "/vsizip/image.tif.zip" + assert info.parameters["in"] == "image.tif.zip" + # File within compressed remote archive + info = pyotb.ReadImageInfo("https://fake.com/archive.tar.gz/image.tif", frozen=True) assert ( info.app.GetParameterValue("in") - == "/vsizip//vsicurl/https://fake.com/image.tif.zip" + == "/vsitar//vsicurl/https://fake.com/archive.tar.gz/image.tif" ) - assert info.parameters["in"] == "https://fake.com/image.tif.zip" + assert info.parameters["in"] == "https://fake.com/archive.tar.gz/image.tif" # Piped curl --> zip --> tiff ziped_tif_urls = ( "https://github.com/OSGeo/gdal/raw/master" @@ -70,8 +76,6 @@ def test_app_input_vsi(): for ziped_tif_url in ziped_tif_urls: info = pyotb.ReadImageInfo(ziped_tif_url) assert info["sizex"] == 20 - # Ensure old way is still working: ExtractROI will raise RuntimeError if a path is malformed - pyotb.Input("/vsicurl/" + SPOT_IMG_URL) def test_img_properties():