senkin: pin LightGBM threads and skip non-applicable datasets

src/methods/senkin_tmp/senkin_tmp_train/config.vsh.yaml

@@ -1,8 +1,18 @@
 __merge__: ../../../api/comp_method_train.yaml
 name: senkin_tmp_train
+info:
+  # The api default test inputs are bmmc_cite/swap (ADT->GEX). senkin predicts protein
+  # (ADT) from RNA (GEX), so point the component test at the 'normal' (GEX->ADT)
+  # direction, otherwise the new modality guard correctly exits non-applicable.
+  test_setup:
+    normal_direction:
+      input_train_mod1: resources_test/task_predict_modality/openproblems_neurips2021/bmmc_cite/normal/train_mod1.h5ad
+      input_train_mod2: resources_test/task_predict_modality/openproblems_neurips2021/bmmc_cite/normal/train_mod2.h5ad
+      input_test_mod1: resources_test/task_predict_modality/openproblems_neurips2021/bmmc_cite/normal/test_mod1.h5ad
 resources:
   - path: script.py
     type: python_script
+  - path: /src/utils/exit_codes.py
 arguments:
   - name: "--n_folds"
     type: integer
@@ -62,3 +72,9 @@ runners:
   - type: nextflow
     directives:
       label: [highmem, hightime, midcpu, gpu]
+# Override the api default (bmmc_cite/swap = ADT->GEX). senkin only supports GEX->ADT.
+test_resources:
+  - type: python_script
+    path: /common/component_tests/run_and_check_output.py
+  - path: /resources_test/task_predict_modality/openproblems_neurips2021/bmmc_cite/normal
+    dest: resources_test/task_predict_modality/openproblems_neurips2021/bmmc_cite/normal

src/methods/senkin_tmp/senkin_tmp_train/script.py

@@ -1,6 +1,7 @@
 import gc
 import logging
 import pickle
+import sys
 
 import anndata as ad
 import numpy as np
@@ -29,9 +30,12 @@
     "nn_epochs": 100,
     "n_tsvd_components": 100,
 }
-meta = {"name": "senkin_tmp"}
+meta = {"name": "senkin_tmp", "resources_dir": "src/methods/senkin_tmp/senkin_tmp_train"}
 ## VIASH END
 
+sys.path.append(meta["resources_dir"])
+from exit_codes import exit_non_applicable
+
 
 def _to_dense(X):
     return X.toarray() if issparse(X) else np.array(X)
@@ -60,6 +64,17 @@ def _split(b):
 adata_prot_train = ad.read_h5ad(par["input_train_mod2"])
 adata_rna_test   = ad.read_h5ad(par["input_test_mod1"])
 
+# senkin is a CITE-seq method that predicts protein (ADT) from RNA (GEX); it treats
+# mod1 as RNA and mod2 as protein. Skip the datasets/directions it cannot handle
+# (e.g. Multiome, or the ADT->GEX swap) instead of running for hours and OOMing.
+_mod1 = adata_rna_train.uns.get("modality")
+_mod2 = adata_prot_train.uns.get("modality")
+if _mod1 != "GEX" or _mod2 != "ADT":
+    exit_non_applicable(
+        f"senkin only supports predicting protein (ADT) from RNA (GEX); "
+        f"got mod1={_mod1!r}, mod2={_mod2!r}."
+    )
+
 adata_rna_train.obs  = _parse_batch(adata_rna_train.obs)
 adata_prot_train.obs = _parse_batch(adata_prot_train.obs)
 adata_rna_test.obs   = _parse_batch(adata_rna_test.obs)
@@ -133,6 +148,15 @@ def _split(b):
 boost_rounds = par["lgbm_boost_rounds"]
 early_stop   = par["lgbm_early_stopping"]
 
+# Pin LightGBM to the allocated cores. Its default (num_threads=0) spawns one thread
+# per core the container *sees* (the whole node) while the job is cgroup-throttled to
+# meta["cpus"], so the threads oversubscribe and thrash -- the same class of slowdown
+# fixed for guanlab in #59. Leave the library default when cpus is unknown (local runs).
+_n_threads = meta.get("cpus")
+if _n_threads:
+    for _p in (lgbm_params_1, lgbm_params_2, lgbm_params_3, lgbm_params_4):
+        _p["num_threads"] = _n_threads
+
 # Protein targets
 Y_prot_raw = _to_dense(adata_prot_train.layers.get("counts", adata_prot_train.X)).astype(np.float64)