Handle empty and "None" string values for run_name and ckpt_dir config parameters (#377)

## Summary
This PR adds defensive checks for empty and "None" string values in
`run_name` and `ckpt_dir` config parameters, along with formatting
improvements across the codebase.

## Changes

### Functional Changes
- **ModelTrainer._setup_ckpt_path()**: Added handling for empty string
(`""`) and string literal `"None"` for both `run_name` and `ckpt_dir`
parameters
- Prevents unexpected behavior when YAML configs have empty values
(e.g., `run_name:` or `ckpt_dir:`)
- Handles edge case where users accidentally set string `"None"` instead
of null

### Test Coverage
- Added test case for empty `run_name` in
`test_model_ckpt_path_duplication`

### Code Formatting
- Removed extra blank lines across multiple files
- Added consistent spacing around operators in f-strings
- Properly wrapped tuple assignments in lightning_modules.py
- Fixed string concatenation in assertion message (inference/utils.py)

## Files Modified
- `sleap_nn/training/model_trainer.py` - defensive checks and formatting
- `sleap_nn/architectures/encoder_decoder.py` - formatting
- `sleap_nn/architectures/unet.py` - formatting
- `sleap_nn/inference/predictors.py` - formatting
- `sleap_nn/inference/topdown.py` - formatting
- `sleap_nn/inference/utils.py` - string formatting fix
- `sleap_nn/tracking/candidates/fixed_window.py` - formatting
- `sleap_nn/tracking/utils.py` - formatting
- `sleap_nn/training/lightning_modules.py` - tuple assignment formatting
- `tests/training/test_model_trainer.py` - added test coverage

## Testing
- Existing tests should pass with these changes
- Added specific test for empty run_name case

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Co-authored-by: Claude <[email protected]>

Author: gitttt-1234

sleap_nn/architectures/encoder_decoder.py

@@ -205,8 +205,8 @@ def __init__(
 
         # Always finish with a pooling block to account for pooling before convs.
         final_pool_dict = OrderedDict()
-        final_pool_dict[f"{self.prefix}{block+1}_last_pool"] = MaxPool2dWithSamePadding(
-            kernel_size=2, stride=2, padding="same"
+        final_pool_dict[f"{self.prefix}{block + 1}_last_pool"] = (
+            MaxPool2dWithSamePadding(kernel_size=2, stride=2, padding="same")
         )
         self.stem_stack.append(nn.Sequential(final_pool_dict))
 

sleap_nn/architectures/unet.py

@@ -124,7 +124,6 @@ def __init__(
             )
             enc_num = len(encoder.encoder_stack)
             if self.middle_block:
-
                 if convs_per_block > 1:
                     # Middle expansion block
                     from sleap_nn.architectures.encoder_decoder import SimpleConvBlock

sleap_nn/inference/predictors.py

@@ -363,7 +363,6 @@ def _predict_generator(self) -> Iterator[Dict[str, np.ndarray]]:
         done = False
 
         try:
-
             with Progress(
                 "{task.description}",
                 BarColumn(),
@@ -378,7 +377,6 @@ def _predict_generator(self) -> Iterator[Dict[str, np.ndarray]]:
                 refresh_per_second=4,  # Change to self.report_rate if needed
                 speed_estimate_period=5,
             ) as progress:
-
                 task = progress.add_task("Predicting...", total=total_frames)
                 last_report = time()
 
@@ -660,7 +658,6 @@ def _initialize_inference_model(self):
             instance_peaks_layer = FindInstancePeaksGroundTruth()
             self.instances_key = True
         else:
-
             max_stride = self.confmap_config.model_config.backbone_config[
                 f"{self.centered_instance_backbone_type}"
             ]["max_stride"]
@@ -1604,7 +1601,6 @@ def _make_labeled_frames_from_generator(
                 ex["pred_peak_values"],
                 ex["orig_size"],
             ):
-
                 if np.isnan(pred_instances).all():
                     continue
                 inst = sio.PredictedInstance.from_numpy(
@@ -2046,7 +2042,6 @@ def _make_labeled_frames_from_generator(
                 ex["pred_peak_values"],
                 ex["instance_scores"],
             ):
-
                 # Loop over instances.
                 predicted_instances = []
                 for pts, confs, score in zip(
@@ -2488,7 +2483,6 @@ def _make_labeled_frames_from_generator(
                 ex["pred_peak_values"],
                 ex["instance_scores"],
             ):
-
                 # Loop over instances.
                 predicted_instances = []
                 for i, (pts, confs, score) in enumerate(

sleap_nn/inference/topdown.py

@@ -799,7 +799,6 @@ def forward(self, batch: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
         batch = self.centroid_crop(batch)
 
         if batch is not None:
-
             if isinstance(self.instance_peaks, FindInstancePeaksGroundTruth):
                 peaks_output.append(self.instance_peaks(batch))
             else:

sleap_nn/inference/utils.py

@@ -47,7 +47,7 @@ def interp1d(x: torch.Tensor, y: torch.Tensor, xnew: torch.Tensor) -> torch.Tens
     v = {}
     eps = torch.finfo(y.dtype).eps
     for name, vec in {"x": x, "y": y, "xnew": xnew}.items():
-        assert len(vec.shape) <= 2, "interp1d: all inputs must be " "at most 2-D."
+        assert len(vec.shape) <= 2, "interp1d: all inputs must be at most 2-D."
         if len(vec.shape) == 1:
             v[name] = vec[None, :]
         else:

sleap_nn/tracking/candidates/fixed_window.py

@@ -134,7 +134,6 @@ def update_tracks(
         """
         add_to_queue = True
         if row_inds is not None and col_inds is not None:
-
             for idx, (row, col) in enumerate(zip(row_inds, col_inds)):
                 current_instances.track_ids[row] = self.current_tracks[col]
                 current_instances.tracking_scores[row] = tracking_scores[idx]

sleap_nn/tracking/utils.py

@@ -133,7 +133,6 @@ def nms_fast(boxes, scores, iou_threshold, target_count=None) -> List[int]:
 
     # keep looping while some indexes still remain in the indexes list
     while len(idxs) > 0:
-
         # we want to add the best box which is the last box in sorted list
         picked_box_idx = idxs[-1]
 

sleap_nn/training/lightning_modules.py

@@ -527,8 +527,9 @@ def forward(self, img):
 
     def training_step(self, batch, batch_idx):
         """Training step."""
-        X, y = torch.squeeze(batch["image"], dim=1), torch.squeeze(
-            batch["confidence_maps"], dim=1
+        X, y = (
+            torch.squeeze(batch["image"], dim=1),
+            torch.squeeze(batch["confidence_maps"], dim=1),
         )
 
         y_preds = self.model(X)["SingleInstanceConfmapsHead"]
@@ -574,8 +575,9 @@ def training_step(self, batch, batch_idx):
 
     def validation_step(self, batch, batch_idx):
         """Validation step."""
-        X, y = torch.squeeze(batch["image"], dim=1), torch.squeeze(
-            batch["confidence_maps"], dim=1
+        X, y = (
+            torch.squeeze(batch["image"], dim=1),
+            torch.squeeze(batch["confidence_maps"], dim=1),
         )
 
         y_preds = self.model(X)["SingleInstanceConfmapsHead"]
@@ -737,8 +739,9 @@ def forward(self, img):
 
     def training_step(self, batch, batch_idx):
         """Training step."""
-        X, y = torch.squeeze(batch["instance_image"], dim=1), torch.squeeze(
-            batch["confidence_maps"], dim=1
+        X, y = (
+            torch.squeeze(batch["instance_image"], dim=1),
+            torch.squeeze(batch["confidence_maps"], dim=1),
         )
 
         y_preds = self.model(X)["CenteredInstanceConfmapsHead"]
@@ -785,8 +788,9 @@ def training_step(self, batch, batch_idx):
 
     def validation_step(self, batch, batch_idx):
         """Perform validation step."""
-        X, y = torch.squeeze(batch["instance_image"], dim=1), torch.squeeze(
-            batch["confidence_maps"], dim=1
+        X, y = (
+            torch.squeeze(batch["instance_image"], dim=1),
+            torch.squeeze(batch["confidence_maps"], dim=1),
         )
 
         y_preds = self.model(X)["CenteredInstanceConfmapsHead"]
@@ -947,8 +951,9 @@ def forward(self, img):
 
     def training_step(self, batch, batch_idx):
         """Training step."""
-        X, y = torch.squeeze(batch["image"], dim=1), torch.squeeze(
-            batch["centroids_confidence_maps"], dim=1
+        X, y = (
+            torch.squeeze(batch["image"], dim=1),
+            torch.squeeze(batch["centroids_confidence_maps"], dim=1),
         )
 
         y_preds = self.model(X)["CentroidConfmapsHead"]
@@ -966,8 +971,9 @@ def training_step(self, batch, batch_idx):
 
     def validation_step(self, batch, batch_idx):
         """Validation step."""
-        X, y = torch.squeeze(batch["image"], dim=1), torch.squeeze(
-            batch["centroids_confidence_maps"], dim=1
+        X, y = (
+            torch.squeeze(batch["image"], dim=1),
+            torch.squeeze(batch["centroids_confidence_maps"], dim=1),
         )
 
         y_preds = self.model(X)["CentroidConfmapsHead"]

sleap_nn/training/model_trainer.py

@@ -226,7 +226,7 @@ def _setup_train_val_labels(
             if skeletons_equal:
                 total_train_lfs += len(train_label)
             else:
-                message = f"The skeletons in the training labels: {index+1} do not match the skeleton in the first training label file."
+                message = f"The skeletons in the training labels: {index + 1} do not match the skeleton in the first training label file."
                 logger.error(message)
                 raise ValueError(message)
 
@@ -291,7 +291,6 @@ def _setup_preprocessing_config(self):
             ):
                 # compute crop size if not provided in config
                 if crop_size is None:
-
                     crop_sz = find_instance_crop_size(
                         labels=train_label,
                         maximum_stride=self.config.model_config.backbone_config[
@@ -358,19 +357,19 @@ def _setup_ckpt_path(self):
         """Setup checkpoint path."""
         # if run_name is None, assign a new dir name
         ckpt_dir = self.config.trainer_config.ckpt_dir
-        if ckpt_dir is None:
+        if ckpt_dir is None or ckpt_dir == "" or ckpt_dir == "None":
             ckpt_dir = "."
             self.config.trainer_config.ckpt_dir = ckpt_dir
         run_name = self.config.trainer_config.run_name
-        if run_name is None:
+        if run_name is None or run_name == "" or run_name == "None":
             sum_train_lfs = sum([len(train_label) for train_label in self.train_labels])
             sum_val_lfs = sum([len(val_label) for val_label in self.val_labels])
             if self._get_trainer_devices() > 1:
-                run_name = f"{self.model_type}.n={sum_train_lfs+sum_val_lfs}"
+                run_name = f"{self.model_type}.n={sum_train_lfs + sum_val_lfs}"
             else:
                 run_name = (
                     datetime.now().strftime("%y%m%d_%H%M%S")
-                    + f".{self.model_type}.n={sum_train_lfs+sum_val_lfs}"
+                    + f".{self.model_type}.n={sum_train_lfs + sum_val_lfs}"
                 )
 
         # If checkpoint path already exists, add suffix to prevent overwriting
@@ -443,7 +442,6 @@ def _verify_model_input_channels(self):
             self.backbone_type == "unet"
             and self.config.model_config.pretrained_backbone_weights is not None
         ):
-
             if self.config.model_config.pretrained_backbone_weights.endswith(".ckpt"):
                 pretrained_backbone_ckpt = torch.load(
                     self.config.model_config.pretrained_backbone_weights,
@@ -648,7 +646,6 @@ def _setup_loggers_callbacks(self, viz_train_dataset, viz_val_dataset):
         loggers = []
         callbacks = []
         if self.config.trainer_config.save_ckpt:
-
             # checkpoint callback
             checkpoint_callback = ModelCheckpoint(
                 save_top_k=self.config.trainer_config.model_ckpt.save_top_k,

tests/training/test_model_trainer.py

@@ -1267,6 +1267,27 @@ def test_model_ckpt_path_duplication(config, caplog, tmp_path, minimal_instance)
     else:
         config.trainer_config.trainer_accelerator = "auto"
 
+    # if run name is empty string
+    cfg_copy = config.copy()
+    OmegaConf.update(
+        cfg_copy,
+        "trainer_config.ckpt_dir",
+        f"{tmp_path}",
+    )
+    OmegaConf.update(
+        cfg_copy,
+        "trainer_config.save_ckpt",
+        True,
+    )
+    OmegaConf.update(cfg_copy, "trainer_config.run_name", "")
+    labels = sio.load_slp(minimal_instance)
+    trainer = ModelTrainer.get_model_trainer_from_config(
+        cfg_copy, train_labels=[labels], val_labels=[labels]
+    )
+
+    trainer.train()
+
+    # use an existing run name
     config_duplicate_ckpt_path = config.copy()
     OmegaConf.update(
         config_duplicate_ckpt_path,