fix: ⚡️ Fix logging, block alignment, tmpdirs.

dacapo/blockwise/predict_worker.py

@@ -165,12 +165,6 @@ def start_worker(
         output_size,
         voxel_size=output_voxel_size,
     )
-    # # use daisy requests to run pipeline
-    # pipeline += gp.DaisyRequestBlocks(
-    #     reference=request,
-    #     roi_map={raw: "read_roi", prediction: "write_roi"},
-    #     num_workers=1,
-    # )
 
     daisy_client = daisy.Client()
 

dacapo/blockwise/relabel_worker.py

@@ -1,6 +1,7 @@
 from glob import glob
 import os
 import sys
+from time import sleep
 import daisy
 from dacapo.compute_context import create_compute_context
 from dacapo.store.array_store import LocalArrayIdentifier
@@ -54,7 +55,14 @@ def start_worker(
             if block is None:
                 break
 
-            relabel_in_block(array_out, nodes, components, block)
+            try:
+                relabel_in_block(array_out, nodes, components, block)
+            except OSError as e:
+                logging.error(
+                    f"Failed to relabel block {block.write_roi}: {e}. Trying again."
+                )
+                sleep(1)
+                relabel_in_block(array_out, nodes, components, block)
 
 
 def relabel_in_block(array_out, old_values, new_values, block):

dacapo/blockwise/segment_worker.py

@@ -167,11 +167,10 @@ def start_worker(
             edges = unique_pairs[non_zero_filter]
             nodes = np.unique(edges)
 
-            print(f"Writing ids to {os.path.join(tmpdir, 'block_%d.npz')}")
             assert os.path.exists(tmpdir)
-            with open(
-                os.path.join(tmpdir, f"block_{block.block_id[1]}.npz"), "wb"
-            ) as f:
+            path = os.path.join(tmpdir, f"block_{block.block_id[1]}.npz")
+            print(f"Writing ids to {path}")
+            with open(path, "wb") as f:
                 np.savez_compressed(
                     f,
                     nodes=nodes,

dacapo/cli.py

@@ -262,6 +262,7 @@ def run_blockwise(
         input_array.voxel_size,
         output_dtype,
         overwrite=overwrite,
+        write_size=write_roi.shape,
     )
 
     _run_blockwise(  # type: ignore
@@ -383,6 +384,7 @@ def segment_blockwise(
         input_array.voxel_size,
         np.uint64,
         overwrite=overwrite,
+        write_size=write_roi.shape,
     )
     print(
         f"Created output array {output_array_identifier.container}:{output_array_identifier.dataset} with ROI {_total_roi}."

dacapo/compute_context/bsub.py

@@ -1,4 +1,7 @@
+import os
+from pathlib import Path
 from .compute_context import ComputeContext
+import daisy
 
 import attr
 
@@ -36,6 +39,10 @@ def device(self):
             return "cpu"
 
     def _wrap_command(self, command):
+        client = daisy.Client()
+        basename = str(
+            Path("./daisy_logs", client.task_id, f"worker_{client.worker_id}")
+        )
         return (
             [
                 "bsub",
@@ -47,10 +54,10 @@ def _wrap_command(self, command):
                 f"num={self.num_gpus}",
                 "-J",
                 "dacapo",
-                # "-o",
-                # f"{run_name}_train.out",
-                # "-e",
-                # f"{run_name}_train.err",
+                "-o",
+                f"{basename}.out",
+                "-e",
+                f"{basename}.err",
             ]
             + (
                 [