vts data errors - 3 telescopes; broken job lib files.

docs/changes/77.bugfix.md

@@ -0,0 +1 @@
+Handle missing highest-numbered telescopes during stereo application and validate trained joblib model artifacts before publishing them.

src/eventdisplay_ml/_model_validation.py

@@ -0,0 +1,53 @@
+"""Isolated validation for serialized model artifacts."""
+
+import sys
+from pathlib import Path
+
+try:
+    import resource
+except ImportError:  # pragma: no cover - resource is unavailable on Windows
+    resource = None
+
+
+def _set_memory_limit(memory_limit_bytes):
+    """Limit virtual memory on Linux before loading third-party model objects."""
+    if resource is not None and sys.platform.startswith("linux"):
+        resource.setrlimit(resource.RLIMIT_AS, (memory_limit_bytes, memory_limit_bytes))
+
+
+def _validate_payload(payload):
+    """Check the minimum structure required by the application model loaders."""
+    if not isinstance(payload, dict):
+        raise TypeError("Model payload must be a dictionary.")
+
+    models = payload.get("models")
+    if not isinstance(models, dict) or not models:
+        raise ValueError("Model payload must contain a non-empty 'models' dictionary.")
+
+    for name, config in models.items():
+        if not isinstance(config, dict) or config.get("model") is None:
+            raise ValueError(f"Model configuration '{name}' does not contain a trained model.")
+
+        model = config["model"]
+        if hasattr(model, "get_booster"):
+            booster = model.get_booster()
+            booster.num_boosted_rounds()
+            booster.num_features()
+
+
+def main():
+    """Load and validate one joblib model file."""
+    if len(sys.argv) != 3:
+        raise SystemExit("Usage: python -m eventdisplay_ml._model_validation FILE MEMORY_BYTES")
+
+    model_path = Path(sys.argv[1])
+    memory_limit_bytes = int(sys.argv[2])
+    _set_memory_limit(memory_limit_bytes)
+
+    from eventdisplay_ml.utils import load_joblib
+
+    _validate_payload(load_joblib(model_path))
+
+
+if __name__ == "__main__":
+    main()

src/eventdisplay_ml/data_processing.py

@@ -1304,8 +1304,9 @@ def _calculate_array_footprint(tel_config, tel_list_matrix):
     n_evt = len(tel_list_matrix)
     footprints = np.full(n_evt, -1.0, dtype=np.float32)
 
-    # Pre-map all telescope positions to a dense array aligned with tel_list_matrix IDs
-    max_id = int(np.nanmax(tel_list_matrix)) if np.any(~np.isnan(tel_list_matrix)) else 0
+    # Size the lookup from the configuration so configured telescopes that are absent
+    # from this event chunk still have a valid slot.
+    max_id = int(np.max(tel_config["tel_ids"]))
     lookup_x = np.zeros(max_id + 1)
     lookup_y = np.zeros(max_id + 1)
     for tid, tx, ty in zip(tel_config["tel_ids"], tel_config["tel_x"], tel_config["tel_y"]):

src/eventdisplay_ml/models.py

@@ -2,6 +2,9 @@
 
 import logging
 import re
+import subprocess
+import sys
+import tempfile
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
@@ -31,10 +34,43 @@
 _EVAL_LOG_E_BINS = 9
 _MIN_WEIGHTED_ENERGY_BIN_EVENTS = 100
 _MAX_REGRESSION_SAMPLE_WEIGHT = 50.0
+_MODEL_VALIDATION_MEMORY_BYTES = 4 * 1024**3
+_MODEL_VALIDATION_TIMEOUT_SECONDS = 120
 
 _logger = logging.getLogger(__name__)
 
 
+def _validate_saved_model(model_path):
+    """Validate a saved model in a memory-limited subprocess."""
+    command = [
+        sys.executable,
+        "-m",
+        "eventdisplay_ml._model_validation",
+        str(model_path),
+        str(_MODEL_VALIDATION_MEMORY_BYTES),
+    ]
+    try:
+        result = subprocess.run(
+            command,
+            capture_output=True,
+            check=False,
+            text=True,
+            timeout=_MODEL_VALIDATION_TIMEOUT_SECONDS,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise RuntimeError(
+            f"Saved model validation timed out after "
+            f"{_MODEL_VALIDATION_TIMEOUT_SECONDS} seconds: {model_path}"
+        ) from exc
+
+    if result.returncode != 0:
+        detail = result.stderr.strip() or result.stdout.strip() or "no diagnostic output"
+        raise RuntimeError(
+            f"Saved model validation failed with exit code {result.returncode}: "
+            f"{model_path}\n{detail}"
+        )
+
+
 def save_models(model_configs):
     """Save trained models to files.
 
@@ -46,7 +82,25 @@ def save_models(model_configs):
         model_configs.get("model_prefix"),
         energy_bin_number=model_configs.get("energy_bin_number"),
     )
-    joblib.dump(model_configs, output_file)
+    output_path = Path(output_file)
+    with tempfile.NamedTemporaryFile(
+        dir=output_path.parent,
+        prefix=f".{output_path.stem}.",
+        suffix=".joblib.gz",
+        delete=False,
+    ) as temporary_file:
+        temporary_path = Path(temporary_file.name)
+
+    try:
+        joblib.dump(model_configs, temporary_path)
+        _logger.info("Validating saved model: %s", temporary_path)
+        _validate_saved_model(temporary_path)
+        temporary_path.replace(output_path)
+    except Exception:
+        temporary_path.unlink(missing_ok=True)
+        raise
+
+    _logger.info("Saved and validated model: %s", output_path)
     utils.log_memory_checkpoint("save_models:end", enabled=memory_profile)
 
 

tests/test_data_helpers.py

@@ -347,8 +347,6 @@ def square_tel_config():
 
 
 def test_calculate_array_footprint_two_tel_returns_distance():
-    # _calculate_array_footprint builds lookup_x[0..max_id] from tel_list_matrix values,
-    # so tel_config must not contain IDs beyond max(tel_list_matrix values).
     tel_config = {
         "tel_ids": np.array([0, 1]),
         "tel_x": np.array([-100.0, 100.0]),
@@ -359,6 +357,14 @@ def test_calculate_array_footprint_two_tel_returns_distance():
     assert result[0] == pytest.approx(200.0)
 
 
+def test_calculate_array_footprint_allows_configured_telescope_absent_from_events(
+    square_tel_config,
+):
+    tel_list_matrix = np.array([[0.0, 1.0, 2.0, np.nan]])
+    result = _calculate_array_footprint(square_tel_config, tel_list_matrix)
+    assert result[0] == pytest.approx(20000.0)
+
+
 def test_calculate_array_footprint_three_tel_returns_positive():
     tel_config = {
         "tel_ids": np.array([0, 1, 2]),

tests/test_models_helpers.py

@@ -50,11 +50,29 @@ def classification_prefix(tmp_path):
 
 
 def test_save_models_writes_expected_joblib(tmp_path):
-    model_configs = {"model_prefix": str(tmp_path / "saved"), "models": {"xgboost": {}}}
+    model_configs = {
+        "model_prefix": str(tmp_path / "saved"),
+        "models": {"xgboost": {"model": "trained-model"}},
+    }
     models.save_models(model_configs)
     assert (tmp_path / "saved.joblib.gz").exists()
 
 
+def test_save_models_rejects_model_that_fails_validation(tmp_path, monkeypatch):
+    model_configs = {
+        "model_prefix": str(tmp_path / "invalid"),
+        "models": {"xgboost": {"model": "trained-model"}},
+    }
+    failed_validation = MagicMock(returncode=1, stderr="validation error", stdout="")
+    monkeypatch.setattr(models.subprocess, "run", lambda *args, **kwargs: failed_validation)
+
+    with pytest.raises(RuntimeError, match="validation error"):
+        models.save_models(model_configs)
+
+    assert not (tmp_path / "invalid.joblib.gz").exists()
+    assert not list(tmp_path.glob(".invalid.*.joblib.gz"))
+
+
 def test_load_models_dispatches_and_rejects_unknown(monkeypatch):
     monkeypatch.setattr(models, "load_regression_models", lambda *args: ("reg", args))
     monkeypatch.setattr(models, "load_classification_models", lambda *args: ("clf", args))