Silence TMVA SOFIE tutorial warnings from third party libraries

Keras and PyTorch still cause some warnings in a few places where they
use deprecated features from their dependencies. We need a mechanism to
silence these warnings to that our tests will run with warnings as
errors, while also making sure we catch when the warnings don't happen
anymore, so we can remove the boilerplate code for warnings silencing.

Author: guitargeek

tutorials/machine_learning/TMVA_SOFIE_Keras.py

@@ -9,11 +9,37 @@
 ### \author Sanjiban Sengupta and Lorenzo Moneta
 
 
+import contextlib
+import warnings
+
 import ROOT
 
 # Enable ROOT in batch mode (same effect as -nodraw)
 ROOT.gROOT.SetBatch(True)
 
+
+@contextlib.contextmanager
+def expect_warning(category, message):
+    """Silence a known third-party warning and raise if it stops firing.
+
+    Notifies us to drop the workaround once the upstream library is fixed.
+    """
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        yield
+    seen = False
+    for w in caught:
+        if issubclass(w.category, category) and message in str(w.message):
+            seen = True
+        else:
+            warnings.warn_explicit(w.message, w.category, w.filename, w.lineno)
+    if not seen:
+        raise RuntimeError(
+            f"Expected {category.__name__} containing {message!r} was not "
+            "emitted. This tutorial's workaround can probably be removed."
+        )
+
+
 # -----------------------------------------------------------------------------
 # Step 1: Create and train a simple Keras model (via embedded Python)
 # -----------------------------------------------------------------------------
@@ -37,7 +63,13 @@
 
 model.compile(loss="mse", optimizer="adam")
 model.fit(x_train, y_train, epochs=3, batch_size=2)
-model.save("KerasModel.keras")
+
+# Keras' internal ``np.array(x)`` (TensorFlow backend) does not yet implement
+# the NumPy 2.0 ``__array__(copy=...)`` signature, so saving the model emits a
+# DeprecationWarning that we cannot fix from user code.
+with expect_warning(DeprecationWarning, "__array__ implementation doesn't accept a copy keyword"):
+    model.save("KerasModel.keras")
+
 model.summary()
 
 # -----------------------------------------------------------------------------

tutorials/machine_learning/TMVA_SOFIE_Keras_HiggsModel.py

@@ -8,6 +8,8 @@
 ### \author Lorenzo Moneta
 
 
+import contextlib
+import warnings
 from os.path import exists
 
 import numpy as np
@@ -16,6 +18,28 @@
 from sklearn.model_selection import train_test_split
 
 
+@contextlib.contextmanager
+def expect_warning(category, message):
+    """Silence a known third-party warning and raise if it stops firing.
+
+    Notifies us to drop the workaround once the upstream library is fixed.
+    """
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        yield
+    seen = False
+    for w in caught:
+        if issubclass(w.category, category) and message in str(w.message):
+            seen = True
+        else:
+            warnings.warn_explicit(w.message, w.category, w.filename, w.lineno)
+    if not seen:
+        raise RuntimeError(
+            f"Expected {category.__name__} containing {message!r} was not "
+            "emitted. This tutorial's workaround can probably be removed."
+        )
+
+
 def CreateModel(nlayers=4, nunits=64):
     input = layers.Input(shape=(7,))
     x = input

tutorials/machine_learning/TMVA_SOFIE_Models.py

@@ -13,21 +13,47 @@
 ### \macro_output
 ### \author Lorenzo Moneta
 
+import contextlib
 import os
+import warnings
 
 import numpy as np
 import ROOT
 from sklearn.model_selection import train_test_split
-from tensorflow.keras.layers import Dense
+from tensorflow.keras.layers import Dense, Input
 from tensorflow.keras.models import Sequential
 from tensorflow.keras.optimizers import Adam
 
+
+@contextlib.contextmanager
+def expect_warning(category, message):
+    """Silence a known third-party warning. Raise if it stops firing.
+
+    Notifies us to drop the workaround once the upstream library is fixed.
+    """
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        yield
+    seen = False
+    for w in caught:
+        if issubclass(w.category, category) and message in str(w.message):
+            seen = True
+        else:
+            warnings.warn_explicit(w.message, w.category, w.filename, w.lineno)
+    if not seen:
+        raise RuntimeError(
+            f"Expected {category.__name__} containing {message!r} was not "
+            "emitted. This tutorial's workaround can probably be removed."
+        )
+
+
 ## generate and train Keras models with different architectures
 
 
 def CreateModel(nlayers=4, nunits=64):
     model = Sequential()
-    model.add(Dense(nunits, activation="relu", input_dim=7))
+    model.add(Input(shape=(7,)))
+    model.add(Dense(nunits, activation="relu"))
     for i in range(1, nlayers):
         model.add(Dense(nunits, activation="relu"))
 
@@ -71,7 +97,11 @@ def PrepareData():
 def TrainModel(model, x, y, name):
     model.fit(x, y, epochs=5, batch_size=50)
     modelFile = name + ".keras"
-    model.save(modelFile)
+    # Keras' internal ``np.array(x)`` (TensorFlow backend) does not yet
+    # implement the NumPy 2.0 ``__array__(copy=...)`` signature, so saving
+    # emits a DeprecationWarning that we cannot fix from user code.
+    with expect_warning(DeprecationWarning, "__array__ implementation doesn't accept a copy keyword"):
+        model.save(modelFile)
     return modelFile
 
 

tutorials/machine_learning/TMVA_SOFIE_ONNX.py

@@ -13,14 +13,38 @@
 ## \author Lorenzo Moneta
 
 
+import contextlib
 import inspect
+import warnings
 
 import numpy as np
 import ROOT
 import torch
 import torch.nn as nn
 
 
+@contextlib.contextmanager
+def expect_warning(category, message):
+    """Silence a known third-party warning and raise if it stops firing.
+
+    Notifies us to drop the workaround once the upstream library is fixed.
+    """
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        yield
+    seen = False
+    for w in caught:
+        if issubclass(w.category, category) and message in str(w.message):
+            seen = True
+        else:
+            warnings.warn_explicit(w.message, w.category, w.filename, w.lineno)
+    if not seen:
+        raise RuntimeError(
+            f"Expected {category.__name__} containing {message!r} was not "
+            "emitted. This tutorial's workaround can probably be removed."
+        )
+
+
 def CreateAndTrainModel(modelName):
 
     model = nn.Sequential(nn.Linear(32, 16), nn.ReLU(), nn.Linear(16, 8), nn.ReLU(), nn.Linear(8, 2), nn.Softmax(dim=1))
@@ -64,7 +88,11 @@ def filtered_kwargs(func, **candidate_kwargs):
     print("calling torch.onnx.export with parameters", kwargs)
 
     try:
-        torch.onnx.export(model, dummy_x, modelFile, **kwargs)
+        # torch.onnx.export (dynamo path) pickles its export program through
+        # copyreg, which still references the deprecated LeafSpec. The warning
+        # is emitted from inside PyTorch and cannot be avoided from user code.
+        with expect_warning(FutureWarning, "isinstance(treespec, LeafSpec)"):
+            torch.onnx.export(model, dummy_x, modelFile, **kwargs)
         print("model exported to ONNX as", modelFile)
         return modelFile
     except TypeError: