Merge pull request #303 from 3DBAG/fix/log-subprocess-commands

Fix/log subprocess commands

Author: balazsdukai

docker/pipeline/bag3d-core.dockerfile

@@ -1,4 +1,4 @@
-FROM 3dgi/3dbag-pipeline-tools:2025.08.17 AS develop
+FROM 3dgi/3dbag-pipeline-tools:2025.08.19 AS develop
 ARG VERSION=develop
 ARG BAG3D_PIPELINE_LOCATION=/opt/3dbag-pipeline
 

docker/pipeline/bag3d-floors-estimation.dockerfile

@@ -1,4 +1,4 @@
-FROM 3dgi/3dbag-pipeline-tools:2025.08.17 AS develop
+FROM 3dgi/3dbag-pipeline-tools:2025.08.19 AS develop
 ARG VERSION=develop
 ARG BAG3D_PIPELINE_LOCATION=/opt/3dbag-pipeline
 

docker/pipeline/bag3d-party-walls.dockerfile

@@ -1,4 +1,4 @@
-FROM 3dgi/3dbag-pipeline-tools:2025.08.17 AS develop
+FROM 3dgi/3dbag-pipeline-tools:2025.08.19 AS develop
 ARG VERSION=develop
 ARG BAG3D_PIPELINE_LOCATION=/opt/3dbag-pipeline
 

packages/common/src/bag3d/common/resources/executables.py

@@ -115,6 +115,7 @@ def execute(
         local_path: Path = None,
         silent=False,
         cwd: str = None,
+        output_logging: str = "STREAM",
     ) -> Tuple[int, str]:
         """Execute a command in a docker container if an image is available, otherwise
         execute with the local executable.
@@ -149,6 +150,9 @@ def execute(
                 mounted on the ``mount_point`` as
                 ``local_path : mount_point/local_path.name``.
             silent: If False, send execution messages to the logger, else do not log.
+            output_logging: The logging mode to use. Supports STREAM, BUFFER, and NONE.
+                STREAM: Stream back logs as they are emitted. BUFFER: Collect and
+                buffer all logs, then emit.
 
         Returns:
              The return code and STDOUT from the command execution.
@@ -200,7 +204,7 @@ def execute(
                 output, return_code = execute_shell_command(
                     shell_command=command.format(**kwargs_with_exe),
                     log=self.logger,
-                    output_logging="STREAM",
+                    output_logging=output_logging,
                     cwd=cwd,
                 )
         if return_code != 0:

packages/common/src/bag3d/common/utils/geodata.py

@@ -74,7 +74,11 @@ def ogrinfo(
     for feature_type in feature_types:
         kwargs = {"xsd": xsd, "dataset": dataset, "feature_type": feature_type}
         return_code, output = gdal.execute(
-            "ogrinfo", command=cmd, kwargs=kwargs, local_path=extract_path
+            "ogrinfo",
+            command=cmd,
+            kwargs=kwargs,
+            local_path=extract_path,
+            output_logging="BUFFER",
         )
         if return_code == 0:
             layername, layerinfo = parse_ogrinfo(output, feature_type)
@@ -235,7 +239,11 @@ def ogr2postgres(
         "dataset": dataset,
     }
     return_code, output = gdal.execute(
-        "ogr2ogr", command=cmd, kwargs=kwargs, local_path=extract_path
+        "ogr2ogr",
+        command=cmd,
+        kwargs=kwargs,
+        local_path=extract_path,
+        output_logging="BUFFER",
     )
     if return_code == 0:
         return postgrestable_metadata(context, new_table)
@@ -267,6 +275,7 @@ def pdal_info(
             command=" ".join(cmd_list),
             local_path=file_path,
             silent=(not verbose),
+            output_logging="BUFFER",
         )
     except Exception as e:
         if "Global encoding WKT flag" in str(e):

packages/core/src/bag3d/core/assets/ahn/metadata.py

@@ -3,7 +3,7 @@
 from functools import partial
 
 import pytz
-from dagster import asset, Output, Field
+from dagster import asset, Output, Field, get_dagster_logger
 from pgutils import PostgresTableIdentifier
 from psycopg.sql import Literal, SQL
 from psycopg.types.json import Jsonb, set_json_dumps
@@ -198,11 +198,12 @@ def compute_load_metadata(
     Returns:
         None
     """
+    logger = get_dagster_logger()
     tile_id = context.partition_key
     conn = context.resources.db_connection.connect
     if not laz_files_ahn.new:
         if not context.op_execution_context.op_execution_context.op_config["force"]:
-            context.log.info(
+            logger.info(
                 f"Metadata for this LAZ tile {tile_id} already exists, "
                 f"skipping computation."
             )
@@ -241,7 +242,7 @@ def compute_load_metadata(
             ST_SetSRID(ST_GeomFromGeoJSON({boundary}), 28992)
         );
         """).format(**query_params)
-    context.log.debug(conn.print_query(query))
+    logger.debug(conn.print_query(query))
     conn.send_query(query)
     # Cannot index the table here, because this is a partitioned assed. This means that
     # this function is called for each partition, which would index the table after
@@ -250,11 +251,12 @@ def compute_load_metadata(
 
 
 def metadata_table_ahn(context, ahn_version: int) -> PostgresTableIdentifier:
+    logger = get_dagster_logger()
     conn = context.resources.db_connection.connect
     new_schema = "ahn"
     create_schema(context, new_schema)
     new_table = PostgresTableIdentifier(new_schema, f"metadata_ahn{ahn_version}")
     query = load_sql(query_params={"new_table": new_table})
-    context.log.info(conn.print_query(query))
+    logger.info(conn.print_query(query))
     conn.send_query(query)
     return new_table

packages/core/tests/test_integration.py

@@ -83,7 +83,7 @@ def test_integration_ahn(database, core_file_store):
                                 "config": {"force_download": False, "check_hash": False}
                             },
                             f"metadata_{ahn_version}": {
-                                "config": {"force": True, "all": True}
+                                "config": {"force": True, "all": True, "verbose": True}
                             },
                             f"lasindex_{ahn_version}": {"config": {"force": True}},
                         }