diff --git a/.gitignore b/.gitignore
index 6a8fbea7b1db968e5d15ebe71ceca7a0697a3b77..5abba97894db9526bdfbbdbbae25497a997a275b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -144,11 +144,10 @@ dmypy.json
# Cython debug symbols
cython_debug/
+# VSCode
+.vscode
+
# PyCharm
-# JetBrains specific template is maintainted in a separate JetBrains.gitignore that can
-# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
-# and can be added to the global gitignore or merged into this file. For a more nuclear
-# option (not recommended) you can uncomment the following to ignore the entire idea folder.
.idea/
# VSCode
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 203b789ac2485afeb0b65490aaad3be1a05dd177..4c01b7b3e6185bcb549f772b64840dd4e44f1e9c 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -45,7 +45,7 @@ pylint:
before_script:
- pip install pylint
script:
- - pylint --max-line-length=120 $PWD/pyotb --disable=too-many-nested-blocks,too-many-locals,too-many-statements,too-few-public-methods,too-many-instance-attributes,too-many-arguments,invalid-name,fixme,too-many-return-statements,too-many-lines,too-many-branches,import-outside-toplevel,wrong-import-position,wrong-import-order,import-error,missing-class-docstring
+ - pylint --max-line-length=120 $PWD/pyotb --disable=too-many-nested-blocks,too-many-locals,too-many-statements,too-few-public-methods,too-many-instance-attributes,too-many-arguments,invalid-name,fixme,too-many-return-statements,too-many-lines,too-many-branches,import-outside-toplevel,wrong-import-position,wrong-import-order,import-error,no-member,no-name-in-module
# ---------------------------------- Documentation ----------------------------------
diff --git a/pyotb/__init__.py b/pyotb/__init__.py
index 187b64a3eab5232464472d46eeca1e1f0ca9f161..634ae623294eb465c6469d3f24c6cfd6240b7932 100644
--- a/pyotb/__init__.py
+++ b/pyotb/__init__.py
@@ -1,8 +1,24 @@
# -*- coding: utf-8 -*-
"""This module provides convenient python wrapping of otbApplications."""
__version__ = "1.5.4"
+from .helpers import find_otb, logger, set_logger_level
+
+otb = find_otb()
from .apps import *
-from .core import App, Output, Input, get_nbchannels, get_pixel_type
-from .functions import * # pylint: disable=redefined-builtin
-from .helpers import logger, set_logger_level
+
+from .core import (
+ Input,
+ Output,
+ get_nbchannels,
+ get_pixel_type
+)
+
+from .functions import ( # pylint: disable=redefined-builtin
+ all,
+ any,
+ where,
+ clip,
+ run_tf_function,
+ define_processing_area
+)
diff --git a/pyotb/apps.py b/pyotb/apps.py
index e5287bbd21a30326b0c9a8341d1d9bb0fa74d3f4..a8dc464ef1f4381e84bdf588e32d2f660a3154c3 100644
--- a/pyotb/apps.py
+++ b/pyotb/apps.py
@@ -4,9 +4,9 @@ import os
import sys
from pathlib import Path
-from .helpers import logger, find_otb
-
-otb = find_otb()
+import otbApplication as otb
+from .helpers import logger
+from .core import OTBObject, DataInterface, RasterInterface, VectorInterface
def get_available_applications(as_subprocess=False):
@@ -34,14 +34,14 @@ def get_available_applications(as_subprocess=False):
cmd_args = [sys.executable, "-c", pycmd]
try:
- import subprocess # pylint: disable=import-outside-toplevel
+ import subprocess
params = {"env": env, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE}
with subprocess.Popen(cmd_args, **params) as p:
logger.debug('Exec "%s \'%s\'"', ' '.join(cmd_args[:-1]), pycmd)
stdout, stderr = p.communicate()
stdout, stderr = stdout.decode(), stderr.decode()
# ast.literal_eval is secure and will raise more handy Exceptions than eval
- from ast import literal_eval # pylint: disable=import-outside-toplevel
+ from ast import literal_eval
app_list = literal_eval(stdout.strip())
assert isinstance(app_list, (tuple, list))
except subprocess.SubprocessError:
@@ -62,18 +62,72 @@ def get_available_applications(as_subprocess=False):
return app_list
-AVAILABLE_APPLICATIONS = get_available_applications(as_subprocess=True)
+class App(OTBObject):
+ """Default parent class of an OTB app. This class should not be used directly without any *Interface."""
-# First core.py call (within __init__ scope)
-from .core import App # pylint: disable=wrong-import-position
+ def __init__(self, appname, *args, **kwargs):
+ """Enables to init an OTB application as a oneliner.
-# This is to enable aliases of Apps, i.e. using apps like `pyotb.AppName(...)` instead of `pyotb.App("AppName", ...)`
-_CODE_TEMPLATE = """
-class {name}(App):
- """ """
- def __init__(self, *args, **kwargs):
- super().__init__('{name}', *args, **kwargs)
-"""
+ Args:
+ appname: name of the application to initialize
+ *args: used for passing application parameters. Can be :
+ - dictionary containing key-arguments enumeration. Useful when a key is python-reserved
+ (e.g. "in") or contains reserved characters such as a point (e.g."mode.extent.unit")
+ - string, App or Output, useful when the user wants to specify the input "in"
+ - list, useful when the user wants to specify the input list 'il'
+ **kwargs: used for passing application parameters.
+ e.g. il=['input1.tif', App_object2, App_object3.out], out='output.tif'
+
+ """
+ super().__init__(appname, *args, **kwargs)
+ self.description = self.app.GetDocLongDescription()
+
+ def find_output_files(self):
+ """Find output files on disk using path found in parameters. Log warning for missing files.
+
+ Returns:
+ list of files found on disk
+
+ """
+ files = []
+ missing = []
+ outputs = (p for p in self.out_param_keys if p in self.parameters)
+ for param in outputs:
+ filename = self.parameters[param]
+ # Remove filename extension
+ if "?" in filename:
+ filename = filename.split("?")[0]
+ path = Path(filename)
+ if path.exists():
+ files.append(path.absolute())
+ else:
+ missing.append(path.absolute())
+ for filename in missing:
+ logger.error("%s: execution seems to have failed, %s does not exist", self.name, filename)
+
+ return tuple(files)
+
+
+class MetaOTBApp(type):
+ """The metaclass (class constructor) to use with any App class child."""
+ def additional_bases(appname): # pylint: disable=bad-mcs-method-argument
+ """Decide which Interface object the app should inherit depending of its output types."""
+ app = otb.Registry_CreateApplicationWithoutLogger(appname)
+ bases = set()
+ keys = app.GetParametersKeys()
+ for k in keys:
+ if app.GetParameterType(k) == otb.ParameterType_OutputImage:
+ bases.add(RasterInterface)
+ elif app.GetParameterType(k) == otb.ParameterType_OutputVectorData:
+ bases.add(VectorInterface)
+ elif app.GetParameterType(k) == otb.ParameterType_OutputFilename:
+ bases.add(DataInterface)
+ # Ensure DataInterface is last in bases in order to avoid erasing RasterInterface's __getitem__ function
+ return tuple(sorted(bases, key=lambda v: v.__name__, reverse=True))
+
+ def __new__(cls, name, bases, classdict):
+ """The function that will instantiate a new class object."""
+ return super().__new__(cls, name, bases + cls.additional_bases(name), classdict)
class OTBTFApp(App):
@@ -89,13 +143,13 @@ class OTBTFApp(App):
"""
if n_sources:
- os.environ['OTB_TF_NSOURCES'] = str(int(n_sources))
+ os.environ["OTB_TF_NSOURCES"] = str(int(n_sources))
else:
# Retrieving the number of `source#.il` parameters
params_dic = {k: v for arg in args if isinstance(arg, dict) for k, v in arg.items()}
- n_sources = len([k for k in params_dic if 'source' in k and k.endswith('.il')])
+ n_sources = len([k for k in params_dic if "source" in k and k.endswith(".il")])
if n_sources >= 1:
- os.environ['OTB_TF_NSOURCES'] = str(n_sources)
+ os.environ["OTB_TF_NSOURCES"] = str(n_sources)
def __init__(self, app_name, *args, n_sources=None, **kwargs):
"""Constructor for an OTBTFApp object.
@@ -110,31 +164,23 @@ class OTBTFApp(App):
self.set_nb_sources(*args, n_sources=n_sources)
super().__init__(app_name, *args, **kwargs)
+# TODO: more specialized classes
+
+
+# This is to enable aliases of Apps, i.e. using apps like `pyotb.AppName(...)` instead of `pyotb.App("AppName", ...)`
+_CODE_TEMPLATE = """
+class {name}(App, metaclass=MetaOTBApp):
+ def __init__(self, *args, **kwargs):
+ super().__init__('{name}', *args, **kwargs)
+"""
+
+AVAILABLE_APPLICATIONS = get_available_applications(as_subprocess=True)
for _app in AVAILABLE_APPLICATIONS:
# Customize the behavior for some OTBTF applications. The user doesn't need to set the env variable
# `OTB_TF_NSOURCES`, it is handled in pyotb
- if _app == 'TensorflowModelServe':
- class TensorflowModelServe(OTBTFApp):
- """Serve a Tensorflow model using OTBTF."""
- def __init__(self, *args, n_sources=None, **kwargs):
- """Constructor for a TensorflowModelServe object."""
- super().__init__('TensorflowModelServe', *args, n_sources=n_sources, **kwargs)
-
- elif _app == 'PatchesExtraction':
- class PatchesExtraction(OTBTFApp):
- """Extract patches using OTBTF."""
- def __init__(self, *args, n_sources=None, **kwargs):
- """Constructor for a PatchesExtraction object."""
- super().__init__('PatchesExtraction', *args, n_sources=n_sources, **kwargs)
-
- elif _app == 'TensorflowModelTrain':
- class TensorflowModelTrain(OTBTFApp):
- """Train a Tensorflow model using OTBTF."""
- def __init__(self, *args, n_sources=None, **kwargs):
- """Constructor for a TensorflowModelTrain object."""
- super().__init__('TensorflowModelTrain', *args, n_sources=n_sources, **kwargs)
-
+ if _app in ("PatchesExtraction", "TensorflowModelTrain", "TensorflowModelServe"):
+ exec(_CODE_TEMPLATE.format(name=_app).replace("(App,", "(OTBTFApp,")) # pylint: disable=exec-used
# Default behavior for any OTB application
else:
exec(_CODE_TEMPLATE.format(name=_app)) # pylint: disable=exec-used
diff --git a/pyotb/core.py b/pyotb/core.py
index 5b9a895b35e342bc16af2b5ab7334b269fa33e49..04e378c262c46d8e8950cfd83775d31f0b5236eb 100644
--- a/pyotb/core.py
+++ b/pyotb/core.py
@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
"""This module is the core of pyotb."""
+from ast import literal_eval
from pathlib import Path
import numpy as np
@@ -8,11 +9,74 @@ import otbApplication as otb
from .helpers import logger
-class otbObject:
- """Base class that gathers common operations for any OTB in-memory raster."""
+class OTBObject:
+ """Base class that gathers common operations for any OTB application."""
_name = ""
- app = None
- output_param = ""
+
+ def __init__(self, appname, *args, frozen=False, quiet=False, **kwargs):
+ """Common constructor for OTB applications. Handles in-memory connection between apps.
+
+ Args:
+ appname: name of the app, e.g. 'BandMath'
+ *args: used for passing application parameters. Can be :
+ - dictionary containing key-arguments enumeration. Useful when a key is python-reserved
+ (e.g. "in") or contains reserved characters such as a point (e.g."mode.extent.unit")
+ - string, App or Output, useful when the user wants to specify the input "in"
+ - list, useful when the user wants to specify the input list 'il'
+ frozen: freeze OTB app in order to use execute() later and avoid blocking process during __init___
+ quiet: whether to print logs of the OTB app
+ **kwargs: used for passing application parameters.
+ e.g. il=['input1.tif', App_object2, App_object3.out], out='output.tif'
+
+ """
+ self.parameters = {}
+ self.appname = appname
+ self.quiet = quiet
+ if quiet:
+ self.app = otb.Registry.CreateApplicationWithoutLogger(appname)
+ else:
+ self.app = otb.Registry.CreateApplication(appname)
+ self.parameters_keys = tuple(self.app.GetParametersKeys())
+ self.out_param_types = dict(self.__get_out_param_types())
+ self.out_param_keys = tuple(self.out_param_types.keys())
+ self.exports_dict = {}
+ if args or kwargs:
+ self.set_parameters(*args, **kwargs)
+ self.frozen = frozen
+ if not frozen:
+ self.execute()
+
+ @property
+ def key_input_image(self):
+ """Get the name of first input image parameter."""
+ types = (otb.ParameterType_InputImage, otb.ParameterType_InputImageList)
+ for key in self.parameters_keys:
+ if self.app.GetParameterType(key) in types:
+ return key
+ return None
+
+ @property
+ def key_input_vector(self):
+ """Get the name of first input image parameter."""
+ types = (otb.ParameterType_InputVectorData, otb.ParameterType_InputVectorDataList)
+ for key in self.parameters_keys:
+ if self.app.GetParameterType(key) in types:
+ return key
+ return None
+
+ @property
+ def key_input_file(self):
+ """Get the name of first input image parameter."""
+ types = (otb.ParameterType_InputFilename, otb.ParameterType_InputFilenameList)
+ for key in self.parameters_keys:
+ if self.app.GetParameterType(key) in types:
+ return key
+ return None
+
+ @property
+ def key_input(self):
+ """Get the name of first input parameter, raster > vector > file."""
+ return self.key_input_image or self.key_input_vector or self.key_input_file
@property
def name(self):
@@ -22,18 +86,300 @@ class otbObject:
user's defined name or appname
"""
- return self._name or self.app.GetName()
+ return self._name or self.appname
@name.setter
- def name(self, val):
+ def name(self, name):
"""Set custom name.
Args:
- val: new name
+ name: new name
+
+ """
+ if isinstance(name, str):
+ self._name = name
+ else:
+ raise TypeError(f"{self.name}: bad type ({type(name)}) for application name, only str is allowed")
+
+ @property
+ def outputs(self):
+ """List of application outputs."""
+ return [getattr(self, key) for key in self.out_param_keys if key in self.parameters]
+
+ def set_parameters(self, *args, **kwargs):
+ """Set some parameters of the app.
+
+ When useful, e.g. for images list, this function appends the parameters
+ instead of overwriting them. Handles any parameters, i.e. in-memory & filepaths
+
+ Args:
+ *args: Can be : - dictionary containing key-arguments enumeration. Useful when a key is python-reserved
+ (e.g. "in") or contains reserved characters such as a point (e.g."mode.extent.unit")
+ - string, App or Output, useful when the user implicitly wants to set the param "in"
+ - list, useful when the user implicitly wants to set the param "il"
+ **kwargs: keyword arguments e.g. il=['input1.tif', oApp_object2, App_object3.out], out='output.tif'
+
+ Raises:
+ Exception: when the setting of a parameter failed
+
+ """
+ parameters = kwargs
+ parameters.update(self.__parse_args(args))
+ # Going through all arguments
+ for key, obj in parameters.items():
+ if key not in self.parameters_keys:
+ raise KeyError(f'{self.name}: unknown parameter name "{key}"')
+ # When the parameter expects a list, if needed, change the value to list
+ if self.__is_key_list(key) and not isinstance(obj, (list, tuple)):
+ obj = [obj]
+ logger.info('%s: argument for parameter "%s" was converted to list', self.name, key)
+ try:
+ # This is when we actually call self.app.SetParameter*
+ self.__set_param(key, obj)
+ except (RuntimeError, TypeError, ValueError, KeyError) as e:
+ raise Exception(
+ f"{self.name}: something went wrong before execution "
+ f"(while setting parameter '{key}' to '{obj}')"
+ ) from e
+ # Update _parameters using values from OtbApplication object
+ otb_params = self.app.GetParameters().items()
+ otb_params = {k: str(v) if isinstance(v, otb.ApplicationProxy) else v for k, v in otb_params}
+ # Save parameters keys, and values as object attributes
+ self.parameters.update({**parameters, **otb_params})
+
+ def execute(self):
+ """Execute and write to disk if any output parameter has been set during init."""
+ logger.debug("%s: run execute() with parameters=%s", self.name, self.parameters)
+ try:
+ self.app.Execute()
+ except (RuntimeError, FileNotFoundError) as e:
+ raise Exception(f"{self.name}: error during during app execution") from e
+ self.frozen = False
+ logger.debug("%s: execution ended", self.name)
+ if any(key in self.parameters for key in self.out_param_keys):
+ self.flush()
+ self.save_objects()
+
+ def write(self, *args, **kwargs):
+ """Write data files to disk.
+
+ Args:
+ args: dict or a filename directly (for raster apps, default output image param will be used)
+ kwargs: key value pairs like parameter=filename
+
+ """
+ if isinstance(self, RasterInterface):
+ self.write_images(*args, **kwargs)
+ return
+ # Any other app, kwargs dict must contain app parameters keys only
+ for arg in args:
+ if isinstance(arg, dict):
+ kwargs.update(arg)
+ self.set_parameters(kwargs)
+ self.flush()
+ self.save_objects()
+
+ def flush(self):
+ """Flush data to disk, this is when WriteOutput is actually called.
+
+ Args:
+ parameters: key value pairs like {parameter: filename}
+ dtypes: optional dict to pass output data types (for rasters)
+
+ """
+ try:
+ logger.debug("%s: flushing data to disk", self.name)
+ self.app.WriteOutput()
+ except RuntimeError:
+ logger.debug("%s: failed with WriteOutput, executing once again with ExecuteAndWriteOutput", self.name)
+ self.app.ExecuteAndWriteOutput()
+
+ def save_objects(self):
+ """Saving app parameters and outputs as attributes, so that they can be accessed with `obj.key`.
+
+ This is useful when the key contains reserved characters such as a point eg "io.out"
+ """
+ for key in self.parameters_keys:
+ if key in dir(OTBObject):
+ continue # skip forbidden attribute since it is already used by the class
+ value = self.parameters.get(key) # basic parameters
+ if value is None:
+ try:
+ value = self.app.GetParameterValue(key) # any other app attribute (e.g. ReadImageInfo results)
+ except RuntimeError:
+ continue # this is when there is no value for key
+ # Convert output param path to Output object
+ if key in self.out_param_keys:
+ value = Output(self, key, value)
+ # Save attribute
+ setattr(self, key, value)
+
+ def summarize(self):
+ """Serialize an object and its pipeline into a dictionary.
+
+ Returns:
+ nested dictionary summarizing the pipeline
+
+ """
+ params = self.parameters
+ for k, p in params.items():
+ # In the following, we replace each parameter which is an otbObject, with its summary.
+ if isinstance(p, OTBObject): # single parameter
+ params[k] = p.summarize()
+ elif isinstance(p, list): # parameter list
+ params[k] = [pi.summarize() if isinstance(pi, OTBObject) else pi for pi in p]
+
+ return {"name": self.name, "parameters": params}
+
+ # Private functions
+ def __parse_args(self, args):
+ """Gather all input arguments in kwargs dict.
+
+ Args:
+ args: the list of arguments passed to set_parameters()
+
+ Returns:
+ a dictionary with the right keyword depending on the object
+
+ """
+ kwargs = {}
+ for arg in args:
+ if isinstance(arg, dict):
+ kwargs.update(arg)
+ elif isinstance(arg, (str, OTBObject)):
+ kwargs.update({self.key_input: arg})
+ elif isinstance(arg, list) and self.__is_key_list(self.key_input):
+ kwargs.update({self.key_input: arg})
+ return kwargs
+
+ def __is_key_list(self, key):
+ """Check if a key of the App is an input parameter list."""
+ return self.app.GetParameterType(key) in (
+ otb.ParameterType_InputImageList,
+ otb.ParameterType_StringList,
+ otb.ParameterType_InputFilenameList,
+ otb.ParameterType_ListView,
+ otb.ParameterType_InputVectorDataList,
+ )
+
+ def __is_key_images_list(self, key):
+ """Check if a key of the App is an input parameter image list."""
+ return self.app.GetParameterType(key) in (otb.ParameterType_InputImageList, otb.ParameterType_InputFilenameList)
+
+ def __get_out_param_types(self):
+ """Get output parameter data type (raster, vector, file).
+
+ Returns:
+ a dictionary like {param_name: param_type}
+
+ """
+ outfile_types = {
+ otb.ParameterType_OutputImage: "raster",
+ otb.ParameterType_OutputVectorData: "vector",
+ otb.ParameterType_OutputFilename: "file",
+ }
+ for k in self.parameters_keys:
+ t = self.app.GetParameterType(k)
+ if t in outfile_types:
+ yield k, outfile_types[t]
+
+ def __set_param(self, key, obj):
+ """Set one parameter, decide which otb.Application method to use depending on target object."""
+ if obj is None or (isinstance(obj, (list, tuple)) and not obj):
+ self.app.ClearValue(key)
+ return
+ if key not in self.parameters_keys:
+ raise Exception(
+ f"{self.name}: parameter '{key}' was not recognized. " f"Available keys are {self.parameters_keys}"
+ )
+ # Single-parameter cases
+ if isinstance(obj, RasterInterface):
+ self.app.ConnectImage(key, obj.app, obj.key_output_image)
+ elif isinstance(obj, otb.Application): # this is for backward comp with plain OTB
+ self.app.ConnectImage(key, obj, get_out_images_param_keys(obj)[0])
+ elif key == "ram": # SetParameterValue in OTB<7.4 doesn't work for ram parameter cf gitlab OTB issue 2200
+ self.app.SetParameterInt("ram", int(obj))
+ elif not isinstance(obj, list): # any other parameters (str, int...)
+ self.app.SetParameterValue(key, obj)
+ # Images list
+ elif self.__is_key_images_list(key):
+ # To enable possible in-memory connections, we go through the list and set the parameters one by one
+ for inp in obj:
+ if isinstance(inp, RasterInterface):
+ self.app.ConnectImage(key, inp.app, inp.key_output_image)
+ elif isinstance(inp, otb.Application): # this is for backward comp with plain OTB
+ self.app.ConnectImage(key, obj, get_out_images_param_keys(inp)[0])
+ else: # here `input` should be an image filepath
+ # Append `input` to the list, do not overwrite any previously set element of the image list
+ self.app.AddParameterStringList(key, inp)
+ # List of any other types (str, int...)
+ else:
+ self.app.SetParameterValue(key, obj)
+
+ def __getattr__(self, name):
+ """This method is called when the default attribute access fails.
+
+ We choose to access the attribute `name` of self.app.
+ Thus, any method of otbApplication can be used transparently on otbObject objects,
+ e.g. SetParameterOutputImagePixelType() or ExportImage() work
+
+ Args:
+ name: attribute name
+
+ Returns:
+ attribute
+
+ Raises:
+ AttributeError: when `name` is not an attribute of self.app
"""
- self._name = val
+ try:
+ res = getattr(self.app, name)
+ return res
+ except AttributeError as e:
+ raise AttributeError(f"{self.name}: could not find attribute `{name}`") from e
+
+ def __hash__(self):
+ """Override the default behaviour of the hash function.
+
+ Returns:
+ self hash
+
+ """
+ return id(self)
+
+ def __str__(self):
+ """Return a nice string representation with object id."""
+ return f"<pyotb.App {self.appname} object id {id(self)}>"
+
+
+class DataInterface:
+ """Object to define interface functions specific to data outputs (e/g. ReadImageInfo)."""
+ @property
+ def data(self):
+ """Expose app's output data values in a dictionary."""
+ keys = (k for k in self.parameters_keys if k not in self.out_param_keys + (self.key_input, 'ram'))
+ dct = {str(k): self[k] for k in keys}
+ empty_obj = ("", None, [], ())
+ return {k: v for k, v in dct.items() if not isinstance(v, otb.ApplicationProxy) and v not in empty_obj}
+
+ @property
+ def key_output_file(self):
+ """Get the name of first output image parameter."""
+ for key in self.parameters_keys:
+ if self.app.GetParameterType(key) == otb.ParameterType_OutputFilename:
+ return key
+ return None
+
+ def __getitem__(self, key):
+ """Access data values using string attributes."""
+ if isinstance(key, str):
+ return self.__dict__.get(key)
+ raise TypeError(f"{self.name}: bad key type ({type(key)}), while accessing object item")
+
+class RasterInterface:
+ """Object to define interface functions specific to raster outputs."""
@property
def dtype(self):
"""Expose the pixel type of an output image using numpy convention.
@@ -43,11 +389,19 @@ class otbObject:
"""
try:
- enum = self.app.GetParameterOutputImagePixelType(self.output_param)
+ enum = self.app.GetParameterOutputImagePixelType(self.key_output_image)
return self.app.ConvertPixelTypeToNumpy(enum)
except RuntimeError:
return None
+ @property
+ def key_output_image(self):
+ """Get the name of first output image parameter."""
+ for key in self.parameters_keys:
+ if self.app.GetParameterType(key) == otb.ParameterType_OutputImage:
+ return key
+ return None
+
@property
def shape(self):
"""Enables to retrieve the shape of a pyotb object using numpy convention.
@@ -56,12 +410,80 @@ class otbObject:
shape: (height, width, bands)
"""
- width, height = self.app.GetImageSize(self.output_param)
- bands = self.app.GetImageNbBands(self.output_param)
+ width, height = self.app.GetImageSize(self.key_output_image)
+ bands = self.app.GetImageNbBands(self.key_output_image)
return (height, width, bands)
- def write(self, *args, filename_extension="", pixel_type=None, **kwargs):
- """Trigger execution, set output pixel type and write the output.
+ def export(self, key=None):
+ """Export a specific output image as numpy array and store it in object's exports_dict.
+
+ Args:
+ key: parameter key to export, if None then the default one will be used
+
+ Returns:
+ the exported numpy array
+
+ """
+ if key is None:
+ key = self.key_output_image
+ if key not in self.exports_dict:
+ self.exports_dict[key] = self.app.ExportImage(key)
+ return self.exports_dict[key]
+
+ def to_numpy(self, key=None, preserve_dtype=True, copy=False):
+ """Export a pyotb object to numpy array.
+
+ Args:
+ key: the output parameter name to export as numpy array
+ preserve_dtype: when set to True, the numpy array is created with the same pixel type as
+ the otbObject first output. Default is True.
+ copy: whether to copy the output array, default is False
+ required to True if preserve_dtype is False and the source app reference is lost
+
+ Returns:
+ a numpy array
+
+ """
+ data = self.export(key)
+ array = data["array"]
+ if preserve_dtype:
+ return array.astype(self.dtype)
+ if copy:
+ return array.copy()
+ return array
+
+ def to_rasterio(self, key=None):
+ """Export image as a numpy array and its metadata compatible with rasterio.
+
+ Args:
+ key: output parameter key to export
+
+ Returns:
+ array : a numpy array in the (bands, height, width) order
+ profile: a metadata dict required to write image using rasterio
+
+ """
+ if key is None:
+ key = self.key_output_image
+ array = self.to_numpy(preserve_dtype=True, copy=False, key=key)
+ array = np.moveaxis(array, 2, 0)
+ proj = self.app.GetImageProjection(key)
+ spacing_x, spacing_y = self.app.GetImageSpacing(key)
+ origin_x, origin_y = self.app.GetImageOrigin(key)
+ # Shift image origin since OTB is giving coordinates of pixel center instead of corners
+ origin_x, origin_y = origin_x - spacing_x / 2, origin_y - spacing_y / 2
+ profile = {
+ "crs": proj,
+ "dtype": array.dtype,
+ "count": array.shape[0],
+ "height": array.shape[1],
+ "width": array.shape[2],
+ "transform": (spacing_x, 0.0, origin_x, 0.0, spacing_y, origin_y),
+ }
+ return array, profile
+
+ def write_images(self, *args, filename_extension="", pixel_type=None, preserve_dtype=False, **kwargs):
+ """Set output pixel type and write the output raster files.
Args:
*args: Can be : - dictionary containing key-arguments enumeration. Useful when a key contains
@@ -75,7 +497,9 @@ class otbObject:
outputs, all outputs are written with this unique type.
Valid pixel types are uint8, uint16, uint32, int16, int32, float, double,
cint16, cint32, cfloat, cdouble. (Default value = None)
+ preserve_dtype: propagate main input pixel type to outputs, in case pixel_type is None
**kwargs: keyword arguments e.g. out='output.tif'
+
"""
# Gather all input arguments in kwargs dict
for arg in args:
@@ -83,102 +507,135 @@ class otbObject:
kwargs.update(arg)
elif isinstance(arg, str) and kwargs:
logger.warning('%s: keyword arguments specified, ignoring argument "%s"', self.name, arg)
- elif isinstance(arg, str):
- kwargs.update({self.output_param: arg})
-
- dtypes = {}
- if isinstance(pixel_type, dict):
- dtypes = {k: parse_pixel_type(v) for k, v in pixel_type.items()}
- elif pixel_type is not None:
- typ = parse_pixel_type(pixel_type)
- if isinstance(self, App):
- dtypes = {key: typ for key in self.output_parameters_keys}
- elif isinstance(self, otbObject):
- dtypes = {self.output_param: typ}
-
+ elif isinstance(arg, str) and self.key_output_image:
+ kwargs.update({self.key_output_image: arg})
+ # Append filename extension to filenames
if filename_extension:
- logger.debug('%s: using extended filename for outputs: %s', self.name, filename_extension)
- if not filename_extension.startswith('?'):
+ logger.debug("%s: using extended filename for outputs: %s", self.name, filename_extension)
+ if not filename_extension.startswith("?"):
filename_extension = "?" + filename_extension
+ for key, value in kwargs.items():
+ if self.out_param_types[key] == 'raster' and '?' not in value:
+ kwargs[key] = value + filename_extension
+ # Manage output pixel types
+ dtypes = {}
+ if pixel_type:
+ if isinstance(pixel_type, str):
+ type_name = self.app.ConvertPixelTypeToNumpy(self.__parse_pixel_type(pixel_type))
+ logger.debug('%s: output(s) will be written with type "%s"', self.name, type_name)
+ for key in kwargs:
+ if self.out_param_types.get(key) == "raster":
+ dtypes[key] = self.__parse_pixel_type(pixel_type)
+ elif isinstance(pixel_type, dict):
+ dtypes = {k: self.__parse_pixel_type(v) for k, v in pixel_type.items()}
+ elif preserve_dtype:
+ self.propagate_dtype() # all outputs will have the same type as the main input raster
+ # Apply parameters
+ for key, output_filename in kwargs.items():
+ if key in dtypes:
+ self.propagate_dtype(key, dtypes[key])
+ self.set_parameters({key: output_filename})
- # Case output parameter was set during App init
- if not kwargs:
- if self.output_param in self.parameters:
- if dtypes:
- self.app.SetParameterOutputImagePixelType(self.output_param, dtypes[self.output_param])
- if filename_extension:
- new_val = self.parameters[self.output_param] + filename_extension
- self.app.SetParameterString(self.output_param, new_val)
- else:
- raise ValueError(f'{self.app.GetName()}: Output parameter is missing.')
+ self.flush()
+ self.save_objects()
- # Parse kwargs
- for key, output_filename in kwargs.items():
- # Stop process if a bad parameter is given
- if key not in self.app.GetParametersKeys():
- raise KeyError(f'{self.app.GetName()}: Unknown parameter key "{key}"')
- # Check if extended filename was not provided twice
- if '?' in output_filename and filename_extension:
- logger.warning('%s: extended filename was provided twice. Using the one found in path.', self.name)
- elif filename_extension:
- output_filename += filename_extension
-
- logger.debug('%s: "%s" parameter is %s', self.name, key, output_filename)
- self.app.SetParameterString(key, output_filename)
+ def propagate_dtype(self, target_key=None, dtype=None):
+ """Propagate a pixel type from main input to every outputs, or to a target output key only.
- if key in dtypes:
- self.app.SetParameterOutputImagePixelType(key, dtypes[key])
+ With multiple inputs (if dtype is not provided), the type of the first input is considered.
+ With multiple outputs (if target_key is not provided), all outputs will be converted to the same pixel type.
- logger.debug('%s: flushing data to disk', self.name)
- try:
- self.app.WriteOutput()
- except RuntimeError:
- logger.debug('%s: failed to simply write output, executing once again then writing', self.name)
- self.app.ExecuteAndWriteOutput()
+ Args:
+ target_key: output param key to change pixel type
+ dtype: data type to use
- def to_numpy(self, preserve_dtype=True, copy=False):
- """Export a pyotb object to numpy array.
+ """
+ if not dtype:
+ param = self.parameters.get(self.key_input_image)
+ if not param:
+ logger.warning("%s: could not propagate pixel type from inputs to output", self.name)
+ return
+ if isinstance(param, (list, tuple)):
+ param = param[0] # first image in "il"
+ try:
+ dtype = get_pixel_type(param)
+ except (TypeError, RuntimeError):
+ logger.warning('%s: unable to identify pixel type of key "%s"', self.name, param)
+ return
+
+ if target_key:
+ keys = [target_key]
+ else:
+ keys = [k for k in self.out_param_keys if self.out_param_types[k] == "raster"]
+ for key in keys:
+ # Set output pixel type
+ self.app.SetParameterOutputImagePixelType(key, dtype)
+
+ def read_values_at_coords(self, row, col, bands=None):
+ """Get pixel value(s) at a given YX coordinates.
Args:
- preserve_dtype: when set to True, the numpy array is created with the same pixel type as
- the otbObject first output. Default is True.
- copy: whether to copy the output array, default is False
- required to True if preserve_dtype is False and the source app reference is lost
+ row: index along Y / latitude axis
+ col: index along X / longitude axis
+ bands: band number, list or slice to fetch values from
Returns:
- a numpy array
+ single numerical value or a list of values for each band
"""
- array = self.app.ExportImage(self.output_param)['array']
- if preserve_dtype:
- return array.astype(self.dtype)
- if copy:
- return array.copy()
- return array
+ channels = []
+ app = OTBObject("PixelValue", self, coordy=row, coordx=col, frozen=False, quiet=True)
+ if bands is not None:
+ if isinstance(bands, int):
+ if bands < 0:
+ bands = self.shape[2] + bands
+ channels = [bands]
+ elif isinstance(bands, slice):
+ channels = self.__channels_list_from_slice(bands)
+ elif not isinstance(bands, list):
+ raise TypeError(f"{self.name}: type '{bands}' cannot be interpreted as a valid slicing")
+ if channels:
+ app.app.Execute()
+ app.set_parameters({"cl": [f"Channel{n+1}" for n in channels]})
+ app.execute()
+ data = literal_eval(app.app.GetParameterString("value"))
+ if len(channels) == 1:
+ return data[0]
+ return data
+
+ def __channels_list_from_slice(self, bands):
+ """Get list of channels to read values at, from a slice."""
+ channels = None
+ start, stop, step = bands.start, bands.stop, bands.step
+ if step is None:
+ step = 1
+ if start is not None and stop is not None:
+ channels = list(range(start, stop, step))
+ elif start is not None and stop is None:
+ channels = list(range(start, self.shape[2], step))
+ elif start is None and stop is not None:
+ channels = list(range(0, stop, step))
+ elif start is None and stop is None:
+ channels = list(range(0, self.shape[2], step))
+ return channels
- def to_rasterio(self):
- """Export image as a numpy array and its metadata compatible with rasterio.
+ @staticmethod
+ def __parse_pixel_type(pixel_type):
+ """Convert one str pixel type to OTB integer enum if necessary.
+
+ Args:
+ pixel_type: pixel type. can be str, int or dict
Returns:
- array : a numpy array in the (bands, height, width) order
- profile: a metadata dict required to write image using rasterio
+ pixel_type integer value
"""
- array = self.to_numpy(preserve_dtype=True, copy=False)
- array = np.moveaxis(array, 2, 0)
- proj = self.app.GetImageProjection(self.output_param)
- spacing_x, spacing_y = self.app.GetImageSpacing(self.output_param)
- origin_x, origin_y = self.app.GetImageOrigin(self.output_param)
- # Shift image origin since OTB is giving coordinates of pixel center instead of corners
- origin_x, origin_y = origin_x - spacing_x / 2, origin_y - spacing_y / 2
- profile = {
- 'crs': proj, 'dtype': array.dtype,
- 'count': array.shape[0], 'height': array.shape[1], 'width': array.shape[2],
- 'transform': (spacing_x, 0.0, origin_x, 0.0, spacing_y, origin_y) # here we force pixel rotation to 0 !
- }
- return array, profile
+ if isinstance(pixel_type, str): # this correspond to 'uint8' etc...
+ return getattr(otb, f"ImagePixelType_{pixel_type}")
+ if isinstance(pixel_type, int):
+ return pixel_type
+ raise ValueError(f"Bad pixel type specification ({pixel_type})")
- # Special methods
def __getitem__(self, key):
"""Override the default __getitem__ behaviour.
@@ -187,47 +644,35 @@ class otbObject:
- slicing, i.e. selecting ROI/bands. For example, selecting first 3 bands: object[:, :, :3]
selecting bands 1, 2 & 5 : object[:, :, [0, 1, 4]]
selecting 1000x1000 subset : object[:1000, :1000]
+ - access pixel value(s) at a specified row, col index
Args:
key: attribute key
Returns:
- attribute or Slicer
+ attribute, pixel values or Slicer
+
"""
# Accessing string attributes
if isinstance(key, str):
return self.__dict__.get(key)
+ # Accessing pixel value(s) using Y/X coordinates
+ if isinstance(key, tuple) and len(key) >= 2:
+ row, col = key[0], key[1]
+ if isinstance(row, int) and isinstance(col, int):
+ if row < 0 or col < 0:
+ raise ValueError(f"{self.name}: can't read pixel value at negative coordinates ({row}, {col})")
+ channels = None
+ if len(key) == 3:
+ channels = key[2]
+ return self.read_values_at_coords(row, col, channels)
# Slicing
if not isinstance(key, tuple) or (isinstance(key, tuple) and (len(key) < 2 or len(key) > 3)):
raise ValueError(f'"{key}"cannot be interpreted as valid slicing. Slicing should be 2D or 3D.')
if isinstance(key, tuple) and len(key) == 2:
# Adding a 3rd dimension
key = key + (slice(None, None, None),)
- (rows, cols, channels) = key
- return Slicer(self, rows, cols, channels)
-
- def __getattr__(self, name):
- """This method is called when the default attribute access fails.
-
- We choose to access the attribute `name` of self.app.
- Thus, any method of otbApplication can be used transparently on otbObject objects,
- e.g. SetParameterOutputImagePixelType() or ExportImage() work
-
- Args:
- name: attribute name
-
- Returns:
- attribute
-
- Raises:
- AttributeError: when `name` is not an attribute of self.app
-
- """
- try:
- res = getattr(self.app, name)
- return res
- except AttributeError as e:
- raise AttributeError(f'{self.name}: could not find attribute `{name}`') from e
+ return Slicer(self, *key)
def __add__(self, other):
"""Overrides the default addition and flavours it with BandMathX.
@@ -236,12 +681,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self + other
+ self + other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return Operation('+', self, other)
+ return Operation("+", self, other)
def __sub__(self, other):
"""Overrides the default subtraction and flavours it with BandMathX.
@@ -250,12 +695,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self - other
+ self - other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return Operation('-', self, other)
+ return Operation("-", self, other)
def __mul__(self, other):
"""Overrides the default subtraction and flavours it with BandMathX.
@@ -264,12 +709,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self * other
+ self * other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return Operation('*', self, other)
+ return Operation("*", self, other)
def __truediv__(self, other):
"""Overrides the default subtraction and flavours it with BandMathX.
@@ -278,12 +723,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self / other
+ self / other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return Operation('/', self, other)
+ return Operation("/", self, other)
def __radd__(self, other):
"""Overrides the default reverse addition and flavours it with BandMathX.
@@ -292,12 +737,12 @@ class otbObject:
other: the other member of the operation
Returns:
- other + self
+ other + self
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return Operation('+', other, self)
+ return Operation("+", other, self)
def __rsub__(self, other):
"""Overrides the default subtraction and flavours it with BandMathX.
@@ -306,12 +751,12 @@ class otbObject:
other: the other member of the operation
Returns:
- other - self
+ other - self
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return Operation('-', other, self)
+ return Operation("-", other, self)
def __rmul__(self, other):
"""Overrides the default multiplication and flavours it with BandMathX.
@@ -320,12 +765,12 @@ class otbObject:
other: the other member of the operation
Returns:
- other * self
+ other * self
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return Operation('*', other, self)
+ return Operation("*", other, self)
def __rtruediv__(self, other):
"""Overrides the default division and flavours it with BandMathX.
@@ -334,12 +779,12 @@ class otbObject:
other: the other member of the operation
Returns:
- other / self
+ other / self
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return Operation('/', other, self)
+ return Operation("/", other, self)
def __abs__(self):
"""Overrides the default abs operator and flavours it with BandMathX.
@@ -348,7 +793,7 @@ class otbObject:
abs(self)
"""
- return Operation('abs', self)
+ return Operation("abs", self)
def __ge__(self, other):
"""Overrides the default greater or equal and flavours it with BandMathX.
@@ -357,12 +802,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self >= other
+ self >= other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return logicalOperation('>=', self, other)
+ return LogicalOperation(">=", self, other)
def __le__(self, other):
"""Overrides the default less or equal and flavours it with BandMathX.
@@ -371,12 +816,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self <= other
+ self <= other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return logicalOperation('<=', self, other)
+ return LogicalOperation("<=", self, other)
def __gt__(self, other):
"""Overrides the default greater operator and flavours it with BandMathX.
@@ -385,12 +830,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self > other
+ self > other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return logicalOperation('>', self, other)
+ return LogicalOperation(">", self, other)
def __lt__(self, other):
"""Overrides the default less operator and flavours it with BandMathX.
@@ -399,12 +844,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self < other
+ self < other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return logicalOperation('<', self, other)
+ return LogicalOperation("<", self, other)
def __eq__(self, other):
"""Overrides the default eq operator and flavours it with BandMathX.
@@ -413,12 +858,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self == other
+ self == other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return logicalOperation('==', self, other)
+ return LogicalOperation("==", self, other)
def __ne__(self, other):
"""Overrides the default different operator and flavours it with BandMathX.
@@ -427,12 +872,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self != other
+ self != other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return logicalOperation('!=', self, other)
+ return LogicalOperation("!=", self, other)
def __or__(self, other):
"""Overrides the default or operator and flavours it with BandMathX.
@@ -441,12 +886,12 @@ class otbObject:
other: the other member of the operation
Returns:
- self || other
+ self || other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return logicalOperation('||', self, other)
+ return LogicalOperation("||", self, other)
def __and__(self, other):
"""Overrides the default and operator and flavours it with BandMathX.
@@ -455,25 +900,16 @@ class otbObject:
other: the other member of the operation
Returns:
- self && other
+ self && other
"""
if isinstance(other, (np.ndarray, np.generic)):
return NotImplemented # this enables to fallback on numpy emulation thanks to __array_ufunc__
- return logicalOperation('&&', self, other)
+ return LogicalOperation("&&", self, other)
# TODO: other operations ?
# e.g. __pow__... cf https://docs.python.org/3/reference/datamodel.html#emulating-numeric-types
- def __hash__(self):
- """Override the default behaviour of the hash function.
-
- Returns:
- self hash
-
- """
- return id(self)
-
def __array__(self):
"""This is called when running np.asarray(pyotb_object).
@@ -499,322 +935,76 @@ class otbObject:
a pyotb object
"""
- if method == '__call__':
+ if method == "__call__":
# Converting potential pyotb inputs to arrays
arrays = []
image_dic = None
for inp in inputs:
if isinstance(inp, (float, int, np.ndarray, np.generic)):
arrays.append(inp)
- elif isinstance(inp, otbObject):
- image_dic = inp.app.ExportImage(self.output_param)
- array = image_dic['array']
+ elif isinstance(inp, OTBObject):
+ if not inp.exports_dict:
+ inp.export()
+ image_dic = inp.exports_dict[inp.key_output_image]
+ array = image_dic["array"]
arrays.append(array)
else:
- print(type(self))
+ logger.debug(type(self))
return NotImplemented
# Performing the numpy operation
result_array = ufunc(*arrays, **kwargs)
result_dic = image_dic
- result_dic['array'] = result_array
+ result_dic["array"] = result_array
- # Importing back to OTB
- app = App('ExtractROI', frozen=True, image_dic=result_dic) # pass the result_dic just to keep reference
+ # Importing back to OTB, pass the result_dic just to keep reference
+ app = InMemoryImage("ExtractROI", image_dic=result_dic, frozen=True, quiet=True)
if result_array.shape[2] == 1:
- app.ImportImage('in', result_dic)
+ app.ImportImage("in", result_dic)
else:
- app.ImportVectorImage('in', result_dic)
+ app.ImportVectorImage("in", result_dic)
app.execute()
return app
return NotImplemented
- def summarize(self):
- """Return a nested dictionary summarizing the otbObject.
- Returns:
- Nested dictionary summarizing the otbObject
+class VectorInterface:
+ """Object to define interface functions specific to vector outputs."""
- """
- params = self.parameters
- for k, p in params.items():
- # In the following, we replace each parameter which is an otbObject, with its summary.
- if isinstance(p, otbObject): # single parameter
- params[k] = p.summarize()
- elif isinstance(p, list): # parameter list
- params[k] = [pi.summarize() if isinstance(pi, otbObject) else pi for pi in p]
+ @property
+ def key_output_vector(self):
+ """Get the name of first output image parameter."""
+ for key in self.parameters_keys:
+ if self.app.GetParameterType(key) == otb.ParameterType_OutputVectorData:
+ return key
+ return None
- return {"name": self.name, "parameters": params}
+class InMemoryImage(OTBObject, RasterInterface):
+ """Class of special in-memory objects like Slicer or Operation."""
-class App(otbObject):
- """Class of an OTB app."""
- def __init__(self, appname, *args, frozen=False, quiet=False,
- preserve_dtype=False, image_dic=None, **kwargs):
- """Enables to init an OTB application as a oneliner. Handles in-memory connection between apps.
+ def __init__(self, appname, *args, image_dic=None, **kwargs):
+ """Default constructor for images in memory.
Args:
- appname: name of the app, e.g. 'Smoothing'
- *args: used for passing application parameters. Can be :
- - dictionary containing key-arguments enumeration. Useful when a key is python-reserved
- (e.g. "in") or contains reserved characters such as a point (e.g."mode.extent.unit")
- - string, App or Output, useful when the user wants to specify the input "in"
- - list, useful when the user wants to specify the input list 'il'
- frozen: freeze OTB app in order to use execute() later and avoid blocking process during __init___
- quiet: whether to print logs of the OTB app
- preserve_dtype: propagate the pixel type from inputs to output. If several inputs, the type of an
- arbitrary input is considered. If several outputs, all will have the same type.
- image_dic: enables to keep a reference to image_dic. image_dic is a dictionary, such as
+ appname: name of the application to initialize (BandMath, ExtractROI...)
+ args: other pyotb app arguments
+ image_dic: enables to keep a reference to input's image_dic. image_dic is a dictionary, such as
the result of app.ExportImage(). Use it when the app takes a numpy array as input.
See this related issue for why it is necessary to keep reference of object:
https://gitlab.orfeo-toolbox.org/orfeotoolbox/otb/-/issues/1824
- **kwargs: used for passing application parameters.
- e.g. il=['input1.tif', App_object2, App_object3.out], out='output.tif'
+ kwargs: other pyotb app keyword arguments
"""
- self.appname = appname
- self.frozen = frozen
- self.quiet = quiet
- self.preserve_dtype = preserve_dtype
self.image_dic = image_dic
- if self.quiet:
- self.app = otb.Registry.CreateApplicationWithoutLogger(appname)
- else:
- self.app = otb.Registry.CreateApplication(appname)
- self.description = self.app.GetDocLongDescription()
- self.output_parameters_keys = self.__get_output_parameters_keys()
- if self.output_parameters_keys:
- self.output_param = self.output_parameters_keys[0]
+ super().__init__(appname, *args, **kwargs)
- self.parameters = {}
- if (args or kwargs):
- self.set_parameters(*args, **kwargs)
- if not self.frozen:
- self.execute()
- def set_parameters(self, *args, **kwargs):
- """Set some parameters of the app.
-
- When useful, e.g. for images list, this function appends the parameters
- instead of overwriting them. Handles any parameters, i.e. in-memory & filepaths
-
- Args:
- *args: Can be : - dictionary containing key-arguments enumeration. Useful when a key is python-reserved
- (e.g. "in") or contains reserved characters such as a point (e.g."mode.extent.unit")
- - string, App or Output, useful when the user implicitly wants to set the param "in"
- - list, useful when the user implicitly wants to set the param "il"
- **kwargs: keyword arguments e.g. il=['input1.tif', oApp_object2, App_object3.out], out='output.tif'
-
- Raises:
- Exception: when the setting of a parameter failed
-
- """
- parameters = kwargs
- parameters.update(self.__parse_args(args))
- # Going through all arguments
- for param, obj in parameters.items():
- if param not in self.app.GetParametersKeys():
- raise Exception(f"{self.name}: parameter '{param}' was not recognized. "
- f"Available keys are {self.app.GetParametersKeys()}")
- # When the parameter expects a list, if needed, change the value to list
- if self.__is_key_list(param) and not isinstance(obj, (list, tuple)):
- parameters[param] = [obj]
- obj = [obj]
- logger.warning('%s: argument for parameter "%s" was converted to list', self.name, param)
- try:
- # This is when we actually call self.app.SetParameter*
- self.__set_param(param, obj)
- except (RuntimeError, TypeError, ValueError, KeyError) as e:
- raise Exception(f"{self.name}: something went wrong before execution "
- f"(while setting parameter '{param}' to '{obj}')") from e
- # Update _parameters using values from OtbApplication object
- otb_params = self.app.GetParameters().items()
- otb_params = {k: str(v) if isinstance(v, otb.ApplicationProxy) else v for k, v in otb_params}
- self.parameters.update({**parameters, **otb_params})
- # Update output images pixel types
- if self.preserve_dtype:
- self.__propagate_pixel_type()
-
- def execute(self):
- """Execute and write to disk if any output parameter has been set during init."""
- logger.debug("%s: run execute() with parameters=%s", self.name, self.parameters)
- try:
- self.app.Execute()
- except (RuntimeError, FileNotFoundError) as e:
- raise Exception(f'{self.name}: error during during app execution') from e
- self.frozen = False
- logger.debug("%s: execution ended", self.name)
- if self.__has_output_param_key():
- logger.debug('%s: flushing data to disk', self.name)
- self.app.WriteOutput()
- self.__save_objects()
-
- def find_output(self):
- """Find output files on disk using path found in parameters.
-
- Returns:
- list of files found on disk
-
- """
- files = []
- missing = []
- outputs = [p for p in self.output_parameters_keys if p in self.parameters]
- for param in outputs:
- filename = self.parameters[param]
- # Remove filename extension
- if '?' in filename:
- filename = filename.split('?')[0]
- path = Path(filename)
- if path.exists():
- files.append(str(path.absolute()))
- else:
- missing.append(str(path.absolute()))
- if missing:
- missing = tuple(missing)
- for filename in missing:
- logger.error("%s: execution seems to have failed, %s does not exist", self.name, filename)
-
- return files
-
- # Private functions
- def __get_output_parameters_keys(self):
- """Get raster output parameter keys.
-
- Returns:
- output parameters keys
- """
- return [param for param in self.app.GetParametersKeys()
- if self.app.GetParameterType(param) == otb.ParameterType_OutputImage]
-
- def __has_output_param_key(self):
- """Check if App has any output parameter key."""
- if not self.output_param:
- return True # apps like ReadImageInfo with no filetype output param still needs to WriteOutput
- types = (otb.ParameterType_OutputFilename, otb.ParameterType_OutputImage, otb.ParameterType_OutputVectorData)
- outfile_params = [param for param in self.app.GetParametersKeys() if self.app.GetParameterType(param) in types]
- return any(key in self.parameters for key in outfile_params)
-
- @staticmethod
- def __parse_args(args):
- """Gather all input arguments in kwargs dict.
-
- Returns:
- a dictionary with the right keyword depending on the object
-
- """
- kwargs = {}
- for arg in args:
- if isinstance(arg, dict):
- kwargs.update(arg)
- elif isinstance(arg, (str, otbObject)):
- kwargs.update({'in': arg})
- elif isinstance(arg, list):
- kwargs.update({'il': arg})
- return kwargs
-
- def __set_param(self, param, obj):
- """Set one parameter, decide which otb.Application method to use depending on target object."""
- if obj is not None:
- # Single-parameter cases
- if isinstance(obj, otbObject):
- self.app.ConnectImage(param, obj.app, obj.output_param)
- elif isinstance(obj, otb.Application): # this is for backward comp with plain OTB
- outparamkey = [param for param in obj.GetParametersKeys()
- if obj.GetParameterType(param) == otb.ParameterType_OutputImage][0]
- self.app.ConnectImage(param, obj, outparamkey)
- elif param == 'ram': # SetParameterValue in OTB<7.4 doesn't work for ram parameter cf gitlab OTB issue 2200
- self.app.SetParameterInt('ram', int(obj))
- elif not isinstance(obj, list): # any other parameters (str, int...)
- self.app.SetParameterValue(param, obj)
- # Images list
- elif self.__is_key_images_list(param):
- # To enable possible in-memory connections, we go through the list and set the parameters one by one
- for inp in obj:
- if isinstance(inp, otbObject):
- self.app.ConnectImage(param, inp.app, inp.output_param)
- elif isinstance(inp, otb.Application): # this is for backward comp with plain OTB
- outparamkey = [param for param in inp.GetParametersKeys() if
- inp.GetParameterType(param) == otb.ParameterType_OutputImage][0]
- self.app.ConnectImage(param, inp, outparamkey)
- else: # here `input` should be an image filepath
- # Append `input` to the list, do not overwrite any previously set element of the image list
- self.app.AddParameterStringList(param, inp)
- # List of any other types (str, int...)
- else:
- self.app.SetParameterValue(param, obj)
-
- def __propagate_pixel_type(self):
- """Propagate the pixel type from inputs to output.
-
- For several inputs, or with an image list, the type of the first input is considered.
- If several outputs, all outputs will have the same type.
-
- """
- pixel_type = None
- for key, param in self.parameters.items():
- if self.__is_key_input_image(key):
- if not param:
- continue
- if isinstance(param, list):
- param = param[0] # first image in "il"
- try:
- pixel_type = get_pixel_type(param)
- type_name = self.app.ConvertPixelTypeToNumpy(pixel_type)
- logger.debug('%s: output(s) will be written with type "%s"', self.name, type_name)
- for out_key in self.output_parameters_keys:
- self.app.SetParameterOutputImagePixelType(out_key, pixel_type)
- return
- except TypeError:
- pass
-
- logger.warning("%s: could not propagate pixel type from inputs to output, no valid input found", self.name)
-
- def __save_objects(self):
- """Saving app parameters and outputs as attributes, so that they can be accessed with `obj.key`.
-
- This is useful when the key contains reserved characters such as a point eg "io.out"
- """
- for key in self.app.GetParametersKeys():
- if key == 'parameters': # skip forbidden attribute since it is already used by the App class
- continue
- value = None
- if key in self.output_parameters_keys: # raster outputs
- value = Output(self, key)
- elif key in self.parameters: # user or default app parameters
- value = self.parameters[key]
- else: # any other app attribute (e.g. ReadImageInfo results)
- try:
- value = self.app.GetParameterValue(key)
- except RuntimeError:
- pass # this is when there is no value for key
- if value is not None:
- setattr(self, key, value)
-
- def __is_key_input_image(self, key):
- """Check if a key of the App is an input parameter image list."""
- return self.app.GetParameterType(key) in (otb.ParameterType_InputImage, otb.ParameterType_InputImageList)
-
- def __is_key_list(self, key):
- """Check if a key of the App is an input parameter list."""
- return self.app.GetParameterType(key) in (otb.ParameterType_InputImageList, otb.ParameterType_StringList,
- otb.ParameterType_InputFilenameList, otb.ParameterType_ListView,
- otb.ParameterType_InputVectorDataList)
-
- def __is_key_images_list(self, key):
- """Check if a key of the App is an input parameter image list."""
- return self.app.GetParameterType(key) in (otb.ParameterType_InputImageList, otb.ParameterType_InputFilenameList)
-
- # Special methods
- def __str__(self):
- """Return a nice string representation with object id."""
- return f'<pyotb.App {self.appname} object id {id(self)}>'
-
-
-class Slicer(App):
+class Slicer(InMemoryImage):
"""Slicer objects i.e. when we call something like raster[:, :, 2] from Python."""
- def __init__(self, x, rows, cols, channels):
+ def __init__(self, obj, rows, cols, channels):
"""Create a slicer object, that can be used directly for writing or inside a BandMath.
It contains :
@@ -822,22 +1012,20 @@ class Slicer(App):
- in case the user only wants to extract one band, an expression such as "im1b#"
Args:
- x: input
+ obj: input
rows: slice along Y / Latitude axis
cols: slice along X / Longitude axis
channels: channels, can be slicing, list or int
"""
- # Initialize the app that will be used for writing the slicer
- self.name = 'Slicer'
-
- self.output_parameter_key = 'out'
- parameters = {'in': x, 'mode': 'extent'}
- super().__init__('ExtractROI', parameters, preserve_dtype=True, frozen=True)
+ super().__init__("ExtractROI", {"in": obj, "mode": "extent"}, quiet=True, frozen=True)
+ self.name = "Slicer"
+ self.rows, self.cols = rows, cols
+ parameters = {}
# Channel slicing
if channels != slice(None, None, None):
# Trigger source app execution if needed
- nb_channels = get_nbchannels(x)
+ nb_channels = get_nbchannels(obj)
self.app.Execute() # this is needed by ExtractROI for setting the `cl` parameter
# if needed, converting int to list
if isinstance(channels, int):
@@ -853,90 +1041,39 @@ class Slicer(App):
elif isinstance(channels, tuple):
channels = list(channels)
elif not isinstance(channels, list):
- raise ValueError(f'Invalid type for channels, should be int, slice or list of bands. : {channels}')
+ raise ValueError(f"Invalid type for channels, should be int, slice or list of bands. : {channels}")
# Change the potential negative index values to reverse index
channels = [c if c >= 0 else nb_channels + c for c in channels]
- parameters.update({'cl': [f'Channel{i + 1}' for i in channels]})
+ parameters.update({"cl": [f"Channel{i + 1}" for i in channels]})
# Spatial slicing
spatial_slicing = False
- # TODO: handle PixelValue app so that accessing value is possible, e.g. raster[120, 200, 0]
# TODO TBD: handle the step value in the slice so that NN undersampling is possible ? e.g. raster[::2, ::2]
if rows.start is not None:
- parameters.update({'mode.extent.uly': rows.start})
+ parameters.update({"mode.extent.uly": rows.start})
spatial_slicing = True
if rows.stop is not None and rows.stop != -1:
- parameters.update(
- {'mode.extent.lry': rows.stop - 1}) # subtract 1 to be compliant with python convention
+ parameters.update({"mode.extent.lry": rows.stop - 1}) # subtract 1 to respect python convention
spatial_slicing = True
if cols.start is not None:
- parameters.update({'mode.extent.ulx': cols.start})
+ parameters.update({"mode.extent.ulx": cols.start})
spatial_slicing = True
if cols.stop is not None and cols.stop != -1:
- parameters.update(
- {'mode.extent.lrx': cols.stop - 1}) # subtract 1 to be compliant with python convention
+ parameters.update({"mode.extent.lrx": cols.stop - 1}) # subtract 1 to respect python convention
spatial_slicing = True
+
# Execute app
- self.set_parameters(**parameters)
+ self.set_parameters(parameters)
self.execute()
-
# These are some attributes when the user simply wants to extract *one* band to be used in an Operation
if not spatial_slicing and isinstance(channels, list) and len(channels) == 1:
self.one_band_sliced = channels[0] + 1 # OTB convention: channels start at 1
- self.input = x
-
-
-class Input(App):
- """Class for transforming a filepath to pyOTB object."""
-
- def __init__(self, filepath):
- """Constructor for an Input object.
-
- Args:
- filepath: raster file path
-
- """
- self.filepath = filepath
- super().__init__('ExtractROI', {'in': self.filepath}, preserve_dtype=True)
+ self.input = obj
+ self.propagate_dtype()
- def __str__(self):
- """Return a nice string representation with input file path."""
- return f'<pyotb.Input object from {self.filepath}>'
-
-
-class Output(otbObject):
- """Class for output of an app."""
-
- def __init__(self, app, output_parameter_key):
- """Constructor for an Output object.
-
- Args:
- app: The pyotb App
- output_parameter_key: Output parameter key
-
- """
- # Keeping the OTB app and the pyotb app
- self.pyotb_app, self.app = app, app.app
- self.parameters = self.pyotb_app.parameters
- self.output_param = output_parameter_key
- self.name = f'Output {output_parameter_key} from {self.app.GetName()}'
-
- def summarize(self):
- """Return the summary of the pipeline that generates the Output object.
-
- Returns:
- Nested dictionary summarizing the pipeline that generates the Output object.
-
- """
- return self.pyotb_app.summarize()
- def __str__(self):
- """Return a nice string representation with object id."""
- return f'<pyotb.Output {self.app.GetName()} object, id {id(self)}>'
-
-
-class Operation(App):
+class Operation(InMemoryImage):
"""Class for arithmetic/math operations done in Python.
Example:
@@ -976,9 +1113,7 @@ class Operation(App):
self.nb_channels = {}
self.fake_exp_bands = []
self.logical_fake_exp_bands = []
-
self.create_fake_exp(operator, inputs, nb_bands=nb_bands)
-
# Transforming images to the adequate im#, e.g. `input1` to "im1"
# creating a dictionary that is like {str(input1): 'im1', 'image2.tif': 'im2', ...}.
# NB: the keys of the dictionary are strings-only, instead of 'complex' objects, to enable easy serialization
@@ -988,20 +1123,18 @@ class Operation(App):
for inp in self.inputs:
if not isinstance(inp, (int, float)):
if str(inp) not in self.im_dic:
- self.im_dic[str(inp)] = f'im{self.im_count}'
+ self.im_dic[str(inp)] = f"im{self.im_count}"
mapping_str_to_input[str(inp)] = inp
self.im_count += 1
- # getting unique image inputs, in the order im1, im2, im3 ...
+ # Getting unique image inputs, in the order im1, im2, im3 ...
self.unique_inputs = [mapping_str_to_input[str_input] for str_input in sorted(self.im_dic, key=self.im_dic.get)]
- self.output_param = 'out'
-
# Computing the BandMath or BandMathX app
self.exp_bands, self.exp = self.get_real_exp(self.fake_exp_bands)
+ # Init app
self.name = f'Operation exp="{self.exp}"'
-
- appname = 'BandMath' if len(self.exp_bands) == 1 else 'BandMathX'
- super().__init__(appname, il=self.unique_inputs, exp=self.exp)
+ appname = "BandMath" if len(self.exp_bands) == 1 else "BandMathX"
+ super().__init__(appname, il=self.unique_inputs, exp=self.exp, quiet=True)
def create_fake_exp(self, operator, inputs, nb_bands=None):
"""Create a 'fake' expression.
@@ -1019,18 +1152,18 @@ class Operation(App):
logger.debug("%s, %s", operator, inputs)
# this is when we use the ternary operator with `pyotb.where` function. The output nb of bands is already known
- if operator == '?' and nb_bands:
+ if operator == "?" and nb_bands:
pass
# For any other operations, the output number of bands is the same as inputs
else:
- if any(isinstance(inp, Slicer) and hasattr(inp, 'one_band_sliced') for inp in inputs):
+ if any(isinstance(inp, Slicer) and hasattr(inp, "one_band_sliced") for inp in inputs):
nb_bands = 1
else:
nb_bands_list = [get_nbchannels(inp) for inp in inputs if not isinstance(inp, (float, int))]
# check that all inputs have the same nb of bands
if len(nb_bands_list) > 1:
if not all(x == nb_bands_list[0] for x in nb_bands_list):
- raise Exception('All images do not have the same number of bands')
+ raise Exception("All images do not have the same number of bands")
nb_bands = nb_bands_list[0]
# Create a list of fake expressions, each item of the list corresponding to one band
@@ -1046,12 +1179,14 @@ class Operation(App):
cond_band = 1
else:
cond_band = band
- fake_exp, corresponding_inputs, nb_channels = self.create_one_input_fake_exp(inp, cond_band,
- keep_logical=True)
+ fake_exp, corresponding_inputs, nb_channels = self.create_one_input_fake_exp(
+ inp, cond_band, keep_logical=True
+ )
# any other input
else:
- fake_exp, corresponding_inputs, nb_channels = self.create_one_input_fake_exp(inp, band,
- keep_logical=False)
+ fake_exp, corresponding_inputs, nb_channels = self.create_one_input_fake_exp(
+ inp, band, keep_logical=False
+ )
fake_exps.append(fake_exp)
# Reference the inputs and nb of channels (only on first pass in the loop to avoid duplicates)
if i == 0 and corresponding_inputs and nb_channels:
@@ -1060,13 +1195,13 @@ class Operation(App):
# Generating the fake expression of the whole operation
if len(inputs) == 1: # this is only for 'abs'
- fake_exp = f'({operator}({fake_exps[0]}))'
+ fake_exp = f"({operator}({fake_exps[0]}))"
elif len(inputs) == 2:
# We create here the "fake" expression. For example, for a BandMathX expression such as '2 * im1 + im2',
# the false expression stores the expression 2 * str(input1) + str(input2)
- fake_exp = f'({fake_exps[0]} {operator} {fake_exps[1]})'
- elif len(inputs) == 3 and operator == '?': # this is only for ternary expression
- fake_exp = f'({fake_exps[0]} ? {fake_exps[1]} : {fake_exps[2]})'
+ fake_exp = f"({fake_exps[0]} {operator} {fake_exps[1]})"
+ elif len(inputs) == 3 and operator == "?": # this is only for ternary expression
+ fake_exp = f"({fake_exps[0]} ? {fake_exps[1]} : {fake_exps[2]})"
self.fake_exp_bands.append(fake_exp)
@@ -1091,7 +1226,7 @@ class Operation(App):
exp_bands.append(one_band_exp)
# Form the final expression (e.g. 'im1b1 + 1; im1b2 + 1')
- exp = ';'.join(exp_bands)
+ exp = ";".join(exp_bands)
return exp_bands, exp
@@ -1116,8 +1251,8 @@ class Operation(App):
"""
# Special case for one-band slicer
- if isinstance(x, Slicer) and hasattr(x, 'one_band_sliced'):
- if keep_logical and isinstance(x.input, logicalOperation):
+ if isinstance(x, Slicer) and hasattr(x, "one_band_sliced"):
+ if keep_logical and isinstance(x.input, LogicalOperation):
fake_exp = x.input.logical_fake_exp_bands[x.one_band_sliced - 1]
inputs = x.input.inputs
nb_channels = x.input.nb_channels
@@ -1128,11 +1263,11 @@ class Operation(App):
nb_channels = x.input.nb_channels
else:
# Add the band number (e.g. replace '<pyotb.App object>' by '<pyotb.App object>b1')
- fake_exp = str(x.input) + f'b{x.one_band_sliced}'
+ fake_exp = str(x.input) + f"b{x.one_band_sliced}"
inputs = [x.input]
nb_channels = {x.input: 1}
- # For logicalOperation, we save almost the same attributes as an Operation
- elif keep_logical and isinstance(x, logicalOperation):
+ # For LogicalOperation, we save almost the same attributes as an Operation
+ elif keep_logical and isinstance(x, LogicalOperation):
fake_exp = x.logical_fake_exp_bands[band - 1]
inputs = x.inputs
nb_channels = x.nb_channels
@@ -1150,23 +1285,22 @@ class Operation(App):
nb_channels = {x: get_nbchannels(x)}
inputs = [x]
# Add the band number (e.g. replace '<pyotb.App object>' by '<pyotb.App object>b1')
- fake_exp = str(x) + f'b{band}'
+ fake_exp = str(x) + f"b{band}"
return fake_exp, inputs, nb_channels
def __str__(self):
- """Return a nice string representation with object id."""
- return f'<pyotb.Operation `{self.operator}` object, id {id(self)}>'
+ """Return a nice string representation with operator and object id."""
+ return f"<pyotb.Operation `{self.operator}` object, id {id(self)}>"
-class logicalOperation(Operation):
+class LogicalOperation(Operation):
"""A specialization of Operation class for boolean logical operations i.e. >, <, >=, <=, ==, !=, `&` and `|`.
The only difference is that not only the BandMath expression is saved (e.g. "im1b1 > 0 ? 1 : 0"), but also the
logical expression (e.g. "im1b1 > 0")
"""
-
def __init__(self, operator, *inputs, nb_bands=None):
"""Constructor for a logicalOperation object.
@@ -1192,39 +1326,94 @@ class logicalOperation(Operation):
"""
# For any other operations, the output number of bands is the same as inputs
- if any(isinstance(inp, Slicer) and hasattr(inp, 'one_band_sliced') for inp in inputs):
+ if any(isinstance(inp, Slicer) and hasattr(inp, "one_band_sliced") for inp in inputs):
nb_bands = 1
else:
nb_bands_list = [get_nbchannels(inp) for inp in inputs if not isinstance(inp, (float, int))]
# check that all inputs have the same nb of bands
if len(nb_bands_list) > 1:
if not all(x == nb_bands_list[0] for x in nb_bands_list):
- raise Exception('All images do not have the same number of bands')
+ raise Exception("All images do not have the same number of bands")
nb_bands = nb_bands_list[0]
# Create a list of fake exp, each item of the list corresponding to one band
for i, band in enumerate(range(1, nb_bands + 1)):
fake_exps = []
for inp in inputs:
- fake_exp, corresponding_inputs, nb_channels = super().create_one_input_fake_exp(inp, band,
- keep_logical=True)
+ fake_exp, corresp_inputs, nb_channels = super().create_one_input_fake_exp(inp, band, keep_logical=True)
fake_exps.append(fake_exp)
# Reference the inputs and nb of channels (only on first pass in the loop to avoid duplicates)
- if i == 0 and corresponding_inputs and nb_channels:
- self.inputs.extend(corresponding_inputs)
+ if i == 0 and corresp_inputs and nb_channels:
+ self.inputs.extend(corresp_inputs)
self.nb_channels.update(nb_channels)
# We create here the "fake" expression. For example, for a BandMathX expression such as 'im1 > im2',
# the logical fake expression stores the expression "str(input1) > str(input2)"
- logical_fake_exp = f'({fake_exps[0]} {operator} {fake_exps[1]})'
+ logical_fake_exp = f"({fake_exps[0]} {operator} {fake_exps[1]})"
# We keep the logical expression, useful if later combined with other logical operations
self.logical_fake_exp_bands.append(logical_fake_exp)
# We create a valid BandMath expression, e.g. "str(input1) > str(input2) ? 1 : 0"
- fake_exp = f'({logical_fake_exp} ? 1 : 0)'
+ fake_exp = f"({logical_fake_exp} ? 1 : 0)"
self.fake_exp_bands.append(fake_exp)
+class FileIO:
+ """Base class of an IO file object."""
+ # TODO: check file exists, create missing directories, ..?
+
+
+class Input(InMemoryImage, FileIO):
+ """Class for transforming a filepath to pyOTB object."""
+
+ def __init__(self, filepath):
+ """Default constructor.
+
+ Args:
+ filepath: the path of an input image
+
+ """
+ super().__init__("ExtractROI", {"in": str(filepath)})
+ self._name = f"Input from {filepath}"
+ self.filepath = Path(filepath)
+ self.propagate_dtype()
+
+ def __str__(self):
+ """Return a nice string representation with file path."""
+ return f"<pyotb.Input object from {self.filepath}>"
+
+
+class Output(FileIO):
+ """Object that behave like a pointer to a specific application output file."""
+
+ def __init__(self, source_app, param_key, filepath=None):
+ """Constructor for an Output object.
+
+ Args:
+ source_app: The pyotb App to store reference from
+ param_key: Output parameter key of the target app
+ filepath: path of the output file (if not in memory)
+
+ """
+ self.source_app = source_app
+ self.param_key = param_key
+ self.filepath = None
+ if filepath:
+ if '?' in filepath:
+ filepath = filepath.split('?')[0]
+ self.filepath = Path(filepath)
+ self.name = f"Output {param_key} from {self.source_app.name}"
+
+ def __str__(self):
+ """Return a nice string representation with source app name and object id."""
+ return f"<pyotb.Output {self.source_app.name} object, id {id(self)}>"
+
+
+def get_out_images_param_keys(app):
+ """Return every output parameter keys of an OTB app."""
+ return [key for key in app.GetParametersKeys() if app.GetParameterType(key) == otb.ParameterType_OutputImage]
+
+
def get_nbchannels(inp):
"""Get the nb of bands of input image.
@@ -1235,15 +1424,14 @@ def get_nbchannels(inp):
number of bands in image
"""
- if isinstance(inp, otbObject):
+ if isinstance(inp, OTBObject):
nb_channels = inp.shape[-1]
else:
- # Executing the app, without printing its log
try:
- info = App("ReadImageInfo", inp, quiet=True)
- nb_channels = info.GetParameterInt("numberbands")
+ info = OTBObject("ReadImageInfo", inp, quiet=True)
+ nb_channels = info.numberbands
except Exception as e: # this happens when we pass a str that is not a filepath
- raise TypeError(f'Could not get the number of channels of `{inp}`. Not a filepath or wrong filepath') from e
+ raise TypeError(f"Could not get the number of channels of `{inp}`. Not a filepath or wrong filepath") from e
return nb_channels
@@ -1259,39 +1447,29 @@ def get_pixel_type(inp):
"""
if isinstance(inp, str):
- # Executing the app, without printing its log
try:
- info = App("ReadImageInfo", inp, quiet=True)
+ info = OTBObject("ReadImageInfo", inp, quiet=True)
except Exception as info_err: # this happens when we pass a str that is not a filepath
- raise TypeError(f'Could not get the pixel type of `{inp}`. Not a filepath or wrong filepath') from info_err
- datatype = info.GetParameterString("datatype") # which is such as short, float...
+ raise TypeError(f"Could not get the pixel type of `{inp}`. Not a filepath or wrong filepath") from info_err
+ datatype = info.datatype # which is such as short, float...
if not datatype:
- raise Exception(f'Unable to read pixel type of image {inp}')
- datatype_to_pixeltype = {'unsigned_char': 'uint8', 'short': 'int16', 'unsigned_short': 'uint16',
- 'int': 'int32', 'unsigned_int': 'uint32', 'long': 'int32', 'ulong': 'uint32',
- 'float': 'float', 'double': 'double'}
+ raise TypeError(f"Unable to read pixel type of image {inp}")
+ datatype_to_pixeltype = {
+ "unsigned_char": "uint8",
+ "short": "int16",
+ "unsigned_short": "uint16",
+ "int": "int32",
+ "unsigned_int": "uint32",
+ "long": "int32",
+ "ulong": "uint32",
+ "float": "float",
+ "double": "double"
+ }
pixel_type = datatype_to_pixeltype[datatype]
- pixel_type = getattr(otb, f'ImagePixelType_{pixel_type}')
- elif isinstance(inp, (otbObject)):
- pixel_type = inp.GetParameterOutputImagePixelType(inp.output_param)
+ pixel_type = getattr(otb, f"ImagePixelType_{pixel_type}")
+ elif isinstance(inp, (OTBObject)):
+ pixel_type = inp.app.GetParameterOutputImagePixelType(inp.key_output_image)
else:
- raise TypeError(f'Could not get the pixel type. Not supported type: {inp}')
+ raise TypeError(f"Could not get the pixel type. Not supported type: {inp}")
return pixel_type
-
-
-def parse_pixel_type(pixel_type):
- """Convert one str pixel type to OTB integer enum if necessary.
-
- Args:
- pixel_type: pixel type. can be str, int or dict
-
- Returns:
- pixel_type integer value
-
- """
- if isinstance(pixel_type, str): # this correspond to 'uint8' etc...
- return getattr(otb, f'ImagePixelType_{pixel_type}')
- if isinstance(pixel_type, int):
- return pixel_type
- raise ValueError(f'Bad pixel type specification ({pixel_type})')
diff --git a/pyotb/functions.py b/pyotb/functions.py
index eee401a3ba6638df6e86f9efe96cc8238c9a4655..4c709bdec16dd678d20bd055de0130c21ba0106e 100644
--- a/pyotb/functions.py
+++ b/pyotb/functions.py
@@ -7,7 +7,8 @@ import textwrap
import uuid
from collections import Counter
-from .core import (otbObject, App, Input, Operation, logicalOperation, get_nbchannels)
+from .apps import ExtractROI, Superimpose
+from .core import RasterInterface, Input, Operation, LogicalOperation, get_nbchannels
from .helpers import logger
@@ -76,8 +77,7 @@ def clip(a, a_min, a_max):
if isinstance(a, str):
a = Input(a)
- res = where(a <= a_min, a_min,
- where(a >= a_max, a_max, a))
+ res = where(a <= a_min, a_min, where(a >= a_max, a_max, a))
return res
@@ -111,25 +111,25 @@ def all(*inputs): # pylint: disable=redefined-builtin
# Checking that all bands of the single image are True
if len(inputs) == 1:
inp = inputs[0]
- if isinstance(inp, logicalOperation):
+ if isinstance(inp, LogicalOperation):
res = inp[:, :, 0]
else:
- res = (inp[:, :, 0] != 0)
+ res = inp[:, :, 0] != 0
for band in range(1, inp.shape[-1]):
- if isinstance(inp, logicalOperation):
+ if isinstance(inp, LogicalOperation):
res = res & inp[:, :, band]
else:
res = res & (inp[:, :, band] != 0)
# Checking that all images are True
else:
- if isinstance(inputs[0], logicalOperation):
+ if isinstance(inputs[0], LogicalOperation):
res = inputs[0]
else:
- res = (inputs[0] != 0)
+ res = inputs[0] != 0
for inp in inputs[1:]:
- if isinstance(inp, logicalOperation):
+ if isinstance(inp, LogicalOperation):
res = res & inp
else:
res = res & (inp != 0)
@@ -166,25 +166,25 @@ def any(*inputs): # pylint: disable=redefined-builtin
# Checking that at least one band of the image is True
if len(inputs) == 1:
inp = inputs[0]
- if isinstance(inp, logicalOperation):
+ if isinstance(inp, LogicalOperation):
res = inp[:, :, 0]
else:
- res = (inp[:, :, 0] != 0)
+ res = inp[:, :, 0] != 0
for band in range(1, inp.shape[-1]):
- if isinstance(inp, logicalOperation):
+ if isinstance(inp, LogicalOperation):
res = res | inp[:, :, band]
else:
res = res | (inp[:, :, band] != 0)
# Checking that at least one image is True
else:
- if isinstance(inputs[0], logicalOperation):
+ if isinstance(inputs[0], LogicalOperation):
res = inputs[0]
else:
- res = (inputs[0] != 0)
+ res = inputs[0] != 0
for inp in inputs[1:]:
- if isinstance(inp, logicalOperation):
+ if isinstance(inp, LogicalOperation):
res = res | inp
else:
res = res | (inp != 0)
@@ -296,11 +296,12 @@ def run_tf_function(func):
# Create and save the model. This is executed **inside an independent process** because (as of 2022-03),
# tensorflow python library and OTBTF are incompatible
- out_savedmodel = os.path.join(tmp_dir, f'tmp_otbtf_model_{uuid.uuid4()}')
+ out_savedmodel = os.path.join(tmp_dir, f"tmp_otbtf_model_{uuid.uuid4()}")
pycmd = get_tf_pycmd(out_savedmodel, channels, scalar_inputs)
cmd_args = [sys.executable, "-c", pycmd]
try:
import subprocess
+
subprocess.run(cmd_args, env=os.environ, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
except subprocess.SubprocessError:
logger.debug("Failed to call subprocess")
@@ -351,9 +352,9 @@ def define_processing_area(*args, window_rule='intersection', pixel_size_rule='m
metadatas = {}
for inp in inputs:
if isinstance(inp, str): # this is for filepaths
- metadata = Input(inp).GetImageMetaData('out')
- elif isinstance(inp, otbObject):
- metadata = inp.GetImageMetaData(inp.output_param)
+ metadata = Input(inp).GetImageMetaData("out")
+ elif isinstance(inp, RasterInterface):
+ metadata = inp.GetImageMetaData(inp.key_output_raster)
else:
raise TypeError(f"Wrong input : {inp}")
metadatas[inp] = metadata
@@ -408,11 +409,11 @@ def define_processing_area(*args, window_rule='intersection', pixel_size_rule='m
for inp in inputs:
try:
params = {
- 'in': inp, 'mode': 'extent', 'mode.extent.unit': 'phy',
+ 'mode': 'extent', 'mode.extent.unit': 'phy',
'mode.extent.ulx': ulx, 'mode.extent.uly': lry, # bug in OTB <= 7.3 :
'mode.extent.lrx': lrx, 'mode.extent.lry': uly, # ULY/LRY are inverted
}
- new_input = App('ExtractROI', params)
+ new_input = ExtractROI(inp, params)
# TODO: OTB 7.4 fixes this bug, how to handle different versions of OTB?
new_inputs.append(new_input)
# Potentially update the reference inputs for later resampling
@@ -453,7 +454,7 @@ def define_processing_area(*args, window_rule='intersection', pixel_size_rule='m
new_inputs = []
for inp in inputs:
if metadatas[inp]['GeoTransform'][1] != pixel_size:
- superimposed = App('Superimpose', inr=reference_input, inm=inp, interpolator=interpolator)
+ superimposed = Superimpose(inr=reference_input, inm=inp, interpolator=interpolator)
new_inputs.append(superimposed)
else:
new_inputs.append(inp)
@@ -478,7 +479,7 @@ def define_processing_area(*args, window_rule='intersection', pixel_size_rule='m
new_inputs = []
for inp in inputs:
if image_sizes[inp] != most_common_image_size:
- superimposed = App('Superimpose', inr=same_size_images[0], inm=inp, interpolator=interpolator)
+ superimposed = Superimpose(inr=same_size_images[0], inm=inp, interpolator=interpolator)
new_inputs.append(superimposed)
else:
new_inputs.append(inp)
diff --git a/pyotb/helpers.py b/pyotb/helpers.py
index 742fee18be4bfbf6951d8e754cb102d470b16b4a..ff30e71a93d7926073195c3be7ad32c503d598b6 100644
--- a/pyotb/helpers.py
+++ b/pyotb/helpers.py
@@ -111,7 +111,7 @@ def set_environment(prefix):
if not prefix.exists():
raise FileNotFoundError(str(prefix))
built_from_source = False
- if not (prefix / 'README').exists():
+ if not (prefix / "README").exists():
built_from_source = True
# External libraries
lib_dir = __find_lib(prefix)
@@ -168,7 +168,7 @@ def __find_lib(prefix=None, otb_module=None):
"""
if prefix is not None:
- lib_dir = prefix / 'lib'
+ lib_dir = prefix / "lib"
if lib_dir.exists():
return lib_dir.absolute()
if otb_module is not None:
@@ -278,16 +278,22 @@ def __suggest_fix_import(error_message, prefix):
if sys.platform == "linux":
if error_message.startswith('libpython3.'):
logger.critical("It seems like you need to symlink or recompile python bindings")
- if sys.executable.startswith('/usr/bin'):
+ if sys.executable.startswith("/usr/bin"):
lib = f"/usr/lib/x86_64-linux-gnu/libpython3.{sys.version_info.minor}.so"
- if which('ctest'):
- logger.critical("To recompile python bindings, use 'cd %s ; source otbenv.profile ; "
- "ctest -S share/otb/swig/build_wrapping.cmake -VV'", prefix)
+ if which("ctest"):
+ logger.critical(
+ "To recompile python bindings, use 'cd %s ; source otbenv.profile ; "
+ "ctest -S share/otb/swig/build_wrapping.cmake -VV'",
+ prefix,
+ )
elif Path(lib).exists():
expect_minor = int(error_message[11])
if expect_minor != sys.version_info.minor:
- logger.critical("Python library version mismatch (OTB was expecting 3.%s) : "
- "a simple symlink may not work, depending on your python version", expect_minor)
+ logger.critical(
+ "Python library version mismatch (OTB was expecting 3.%s) : "
+ "a simple symlink may not work, depending on your python version",
+ expect_minor,
+ )
target_lib = f"{prefix}/lib/libpython3.{expect_minor}.so.rh-python3{expect_minor}-1.0"
logger.critical("Use 'ln -s %s %s'", lib, target_lib)
else:
diff --git a/tests/test_core.py b/tests/test_core.py
index abefc80c7c5a03c06360e900d2e7dabee4e116b3..2e29beefb4a6fb413f532a4a8762bb8849f93874 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -33,6 +33,14 @@ def test_operation():
assert op.exp == "((im1b1 / 255) * 128);((im1b2 / 255) * 128);((im1b3 / 255) * 128);((im1b4 / 255) * 128)"
+def test_operation_write_dtype():
+ op = INPUT / 255 * 128
+ assert op.dtype == "float32"
+ op.write("test.tif", pixel_type="uint8")
+ assert Path("test.tif").exists()
+ assert op.dtype == "uint8"
+
+
def test_sum_bands():
# Sum of bands
summed = sum(INPUT[:, :, b] for b in range(INPUT.shape[-1]))
diff --git a/tests/test_numpy.py b/tests/test_numpy.py
index 5b1dd048cb73d2a9c904c3b937c6d60d8b8520c3..878047338fbc6e52e7592b5411e950a5bbea34e4 100644
--- a/tests/test_numpy.py
+++ b/tests/test_numpy.py
@@ -31,7 +31,7 @@ def test_convert_to_array():
def test_add_noise_array():
white_noise = np.random.normal(0, 50, size=INPUT.shape)
noisy_image = INPUT + white_noise
- assert isinstance(noisy_image, pyotb.otbObject)
+ assert isinstance(noisy_image, pyotb.core.OTBObject)
assert noisy_image.shape == INPUT.shape
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index c2d8f0e910851f5412f044c5d4ef887613af57a0..0cf5857d16cb4d4be7d98853bcd39fb862925361 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -11,7 +11,7 @@ OTBAPPS_BLOCKS = [
# lambda inp: pyotb.ExtractROI({"in": inp, "startx": 10, "starty": 10, "sizex": 50, "sizey": 50}),
lambda inp: pyotb.ManageNoData({"in": inp, "mode": "changevalue"}),
lambda inp: pyotb.DynamicConvert({"in": inp}),
- lambda inp: pyotb.Mosaic({"il": [inp]}),
+ #lambda inp: pyotb.Mosaic({"il": [inp]}),
lambda inp: pyotb.BandMath({"il": [inp], "exp": "im1b1 + 1"}),
lambda inp: pyotb.BandMathX({"il": [inp], "exp": "im1"})
]
@@ -20,7 +20,7 @@ PYOTB_BLOCKS = [
lambda inp: 1 + abs(inp) * 2,
lambda inp: inp[:80, 10:60, :],
]
-PIPELINES_LENGTH = [1, 2, 3]
+PIPELINES_LENGTH = [2, 3]
ALL_BLOCKS = PYOTB_BLOCKS + OTBAPPS_BLOCKS
FILEPATH = os.environ["TEST_INPUT_IMAGE"]