feat: add decorator to check if las file is written correctly

fix: tests after adding decorator

fix: tests using check_las

fix: review

fix: review

fix: tests

fix: tests and doc

fix: tests

Author: nlenglet-ign

CHANGELOG.md

@@ -1,3 +1,6 @@
+# dev
+- [new] add check_las and use it when needed
+
 # 1.17.0
 - [new] add function to launch a pdal pipeline on a Las file
 

pdaltools/check_las.py

@@ -0,0 +1,62 @@
+"""Check that LAS/LAZ files can be read by PDAL.
+
+Provides a one-shot check, a retry variant (for files still being written), and a
+decorator that validates an output path after a wrapped function returns.
+"""
+
+import functools
+import inspect
+import os
+import time
+
+import pdal
+
+
+def check_pdal_can_open_file(filepath: str) -> bool:
+    try:
+        if not os.path.exists(filepath):
+            raise FileNotFoundError(f"File {filepath} does not exist.")
+        pipeline = pdal.Reader.las(filename=filepath).pipeline()
+        pipeline.execute()
+        return True
+    except RuntimeError as e:
+        if e.__str__().startswith("readers.las:"):
+            print(f"Pdal could not read {filepath} due to error: {e}")
+            return False
+        else:
+            raise e
+
+
+def check_pdal_can_open_file_with_retry(filepath: str, delay: int) -> bool:
+    if check_pdal_can_open_file(filepath):
+        return True
+    else:
+        print(f"New attempt in {delay} seconds.")
+        time.sleep(delay)
+        return check_pdal_can_open_file(filepath)
+
+
+def check_pdal_can_open_file_with_retry_decorator(delay: int, filepath: str | None = None):
+    def decorator(fn):
+        sig = inspect.signature(fn)
+
+        # Extract the LAS path from the wrapped call: bind args/kwargs to the
+        # function signature, apply defaults if needed, then return the parameter
+        # named by `filepath` (e.g. "output_file" or the first positional arg).
+        def resolve_filepath(*args, **kwargs) -> str:
+            bound = sig.bind_partial(*args, **kwargs)
+            if filepath not in bound.arguments:
+                bound.apply_defaults()
+            return str(bound.arguments[filepath])
+
+        @functools.wraps(fn)
+        def wrapper(*args, **kwargs):
+            result = fn(*args, **kwargs)
+            resolved_filepath = resolve_filepath(*args, **kwargs)
+            if not check_pdal_can_open_file_with_retry(resolved_filepath, delay):
+                raise RuntimeError(f"Pdal could not read {resolved_filepath} after retry.")
+            return result
+
+        return wrapper
+
+    return decorator

pdaltools/color.py

@@ -6,6 +6,7 @@
 import pdal
 
 import pdaltools.las_info as las_info
+from pdaltools.check_las import check_pdal_can_open_file_with_retry_decorator
 from pdaltools.download_image import download_image
 
 
@@ -31,6 +32,7 @@ def match_min_max_with_pixel_size(min_d: float, max_d: float, pixel_per_meter: f
     return min_d, max_d
 
 
+@check_pdal_can_open_file_with_retry_decorator(delay=10, filepath="output_file")
 def color_from_stream(
     input_file: str,
     output_file: str,
@@ -156,6 +158,7 @@ def color_from_stream(
     return tmp_ortho, tmp_ortho_irc
 
 
+@check_pdal_can_open_file_with_retry_decorator(delay=10, filepath="output_file")
 def color_from_files(
     input_file: str,
     output_file: str,

pdaltools/las_add_extra_dims_from_las.py

@@ -15,6 +15,8 @@
 import laspy
 import numpy as np
 
+from pdaltools.check_las import check_pdal_can_open_file_with_retry_decorator
+
 _LAS_SUFFIXES = frozenset({".las", ".laz"})
 logger = logging.getLogger(__name__)
 # ANSI for terminal emphasis (ignored by non-TTY handlers in most setups).
@@ -160,6 +162,7 @@ def _dims_to_copy(base: laspy.LasData, source: laspy.LasData, dimensions: Option
     return [d for d in requested if d in missing]
 
 
+@check_pdal_can_open_file_with_retry_decorator(delay=10, filepath="output_las")
 def add_extra_dims_from_las(
     base_las: Path | str,
     source_las: Path | str,

pdaltools/standardize_format.py

@@ -15,6 +15,7 @@
 
 import pdal
 
+from pdaltools.check_las import check_pdal_can_open_file_with_retry_decorator
 from pdaltools.las_rename_dimension import rename_dimension
 from pdaltools.unlock_file import copy_and_hack_decorator
 
@@ -79,16 +80,13 @@ def get_writer_parameters(new_parameters: Dict) -> Dict:
 
 
 @copy_and_hack_decorator
+@check_pdal_can_open_file_with_retry_decorator(delay=10, filepath="output_file")
 def standardize(
-    input_file: str, 
-    output_file: str, 
-    params_from_parser: Dict, 
-    classes_to_remove: List = [], 
-    rename_dims: List = []
+    input_file: str, output_file: str, params_from_parser: Dict, classes_to_remove: List = [], rename_dims: List = []
 ) -> None:
     """
     Standardize a LAS/LAZ file with improved error handling and resource management.
-    
+
     Args:
         input_file: Input file path
         output_file: Output file path

test/data/check_las/Semis_2022_0906_6665_LA93_IGN69_nok.laz

@@ -0,0 +1,71 @@
+"""Check if a LAS file is written correcty and can be opened by PDAL"""
+
+import os
+import shutil
+
+import pytest
+
+from pdaltools.check_las import (
+    check_pdal_can_open_file,
+    check_pdal_can_open_file_with_retry,
+    check_pdal_can_open_file_with_retry_decorator,
+)
+
+TEST_PATH = os.path.dirname(os.path.abspath(__file__))
+INPUT_DIR = os.path.join(TEST_PATH, "data/check_las")
+
+
+def test_check_pdal_can_open_file():
+    filepath = os.path.join(INPUT_DIR, "Semis_2022_0906_6665_LA93_IGN69_ok.laz")
+    assert check_pdal_can_open_file(filepath)
+
+
+def test_check_pdal_can_open_file_nok():
+    filepath = os.path.join(INPUT_DIR, "Semis_2022_0906_6665_LA93_IGN69_nok.laz")
+    assert not check_pdal_can_open_file(filepath)
+
+
+def test_check_pdal_can_open_file_with_retry():
+    filepath = os.path.join(INPUT_DIR, "Semis_2022_0906_6665_LA93_IGN69_ok.laz")
+    assert check_pdal_can_open_file_with_retry(filepath, 1)
+
+
+def test_check_pdal_can_open_file_with_retry_nok():
+    filepath = os.path.join(INPUT_DIR, "Semis_2022_0906_6665_LA93_IGN69_nok.laz")
+    assert not check_pdal_can_open_file_with_retry(filepath, 1)
+
+
+@check_pdal_can_open_file_with_retry_decorator(delay=0, filepath="filepath")  # or mock time.sleep
+def echo(filepath):
+    return filepath
+
+
+def test_decorator_passes_through_on_ok():
+    filepath = os.path.join(INPUT_DIR, "Semis_2022_0906_6665_LA93_IGN69_ok.laz")
+    assert echo(filepath) == filepath
+
+
+def test_decorator_raises_on_nok():
+    filepath = os.path.join(INPUT_DIR, "Semis_2022_0906_6665_LA93_IGN69_nok.laz")
+    with pytest.raises(RuntimeError):
+        echo(filepath)
+
+
+@check_pdal_can_open_file_with_retry_decorator(delay=0, filepath="output_file")
+def copy_las(input_file, output_file):
+    shutil.copy(input_file, output_file)
+
+
+def test_decorator_post_write_passes_on_ok(tmp_path):
+    input_file = os.path.join(INPUT_DIR, "Semis_2022_0906_6665_LA93_IGN69_ok.laz")
+    output_file = tmp_path / "output.laz"
+    copy_las(input_file, output_file)
+    print(output_file)
+    assert output_file.is_file()
+
+
+def test_decorator_post_write_raises_on_nok(tmp_path):
+    input_file = os.path.join(INPUT_DIR, "Semis_2022_0906_6665_LA93_IGN69_nok.laz")
+    output_file = tmp_path / "output.laz"
+    with pytest.raises(RuntimeError):
+        copy_las(input_file, output_file)