Add example partial filter

ml_peg/analysis/surfaces/elemental_slab_oxygen_adsorption/analyse_elemental_slab_oxygen_adsorption.py

@@ -28,14 +28,24 @@
     METRICS_CONFIG_PATH
 )
 
+
 # Extract system metadata from mock calculation (filenames)
-SYSTEM_INFO = get_struct_info(
-    calc_path=CALC_PATH,
-    glob_pattern="*.xyz",
-    index="0",
-    include_filenames=True,
-    out_path=OUT_PATH,
-)
+def get_info() -> dict[str, list]:
+    """
+    Get and write out metadata for each system.
+
+    Returns
+    -------
+    dict[str, list]
+        Metadata for each system.
+    """
+    return get_struct_info(
+        calc_path=CALC_PATH,
+        glob_pattern="*.xyz",
+        index="1",
+        include_filenames=True,
+        out_path=OUT_PATH,
+    )
 
 
 def compute_adsorption_energy(
@@ -68,7 +78,7 @@ def compute_adsorption_energy(
     x_label="Predicted adsorption energy / eV",
     y_label="Reference adsorption energy / eV",
     hoverdata={
-        "System": SYSTEM_INFO["filenames"],
+        "System": get_info()["filenames"],
     },
 )
 def adsorption_energies() -> dict[str, list]:
@@ -117,7 +127,6 @@ def adsorption_energies() -> dict[str, list]:
     return results
 
 
-@pytest.fixture
 def adsorption_mae(adsorption_energies) -> dict[str, float]:
     """
     Get mean absolute error for adsorption energies.
@@ -134,7 +143,7 @@ def adsorption_mae(adsorption_energies) -> dict[str, float]:
     """
     results = {}
     for model_name in MODELS:
-        if adsorption_energies[model_name]:
+        if adsorption_energies.get(model_name):
             results[model_name] = mae(
                 adsorption_energies["ref"], adsorption_energies[model_name]
             )
@@ -143,29 +152,46 @@ def adsorption_mae(adsorption_energies) -> dict[str, float]:
     return results
 
 
+def get_metrics(adsorption_energies: dict[str, float]) -> dict[str, dict]:
+    """
+    Get all metrics.
+
+    Parameters
+    ----------
+    adsorption_energies
+        Dictionary of all reference and predicted adsorption energies.
+
+    Returns
+    -------
+    dict[str, dict]
+        Metric names and values for all models.
+    """
+    return {
+        "MAE": adsorption_mae(adsorption_energies),
+    }
+
+
 @pytest.fixture
 @build_table(
     filename=OUT_PATH / "elemental_slab_oxygen_adsorption_metrics_table.json",
     metric_tooltips=DEFAULT_TOOLTIPS,
     thresholds=DEFAULT_THRESHOLDS,
 )
-def metrics(adsorption_mae: dict[str, float]) -> dict[str, dict]:
+def metrics(adsorption_energies: dict[str, list[float]]) -> dict[str, dict]:
     """
-    Get all metrics.
+    Get all GMTKN55 metrics.
 
     Parameters
     ----------
-    adsorption_mae
-        Mean absolute errors for all models.
+    adsorption_energies
+        Dictionary of reference and predicted adsorption energies.
 
     Returns
     -------
     dict[str, dict]
         Metric names and values for all models.
     """
-    return {
-        "MAE": adsorption_mae,
-    }
+    return get_metrics(adsorption_energies)
 
 
 def test_elemental_slab_oxygen_adsorption(metrics: dict[str, dict]) -> None:

ml_peg/analysis/utils/utils.py

@@ -158,7 +158,7 @@ def mae(ref: list, prediction: list) -> float:
     float
         Mean absolute error.
     """
-    if np.isnan(np.sum(prediction)):
+    if (np.array(prediction) == None).any() or np.isnan(np.sum(prediction)):  # noqa: E711
         return np.nan
     return mean_absolute_error(ref, prediction)
 
@@ -179,7 +179,7 @@ def rmse(ref: list, prediction: list) -> float:
     float
         Root mean squared error.
     """
-    if np.isnan(np.sum(prediction)):
+    if (np.array(prediction) == None).any() or np.isnan(np.sum(prediction)):  # noqa: E711
         return np.nan
     return np.sqrt(mean_squared_error(ref, prediction))
 

ml_peg/app/surfaces/elemental_slab_oxygen_adsorption/app_elemental_slab_oxygen_adsorption.py

@@ -2,21 +2,24 @@
 
 from __future__ import annotations
 
+from copy import deepcopy
+import warnings
+
 from dash import Dash
 from dash.html import Div
+import numpy as np
 
+from ml_peg.analysis.surfaces.elemental_slab_oxygen_adsorption.analyse_elemental_slab_oxygen_adsorption import (  # noqa: E501
+    get_metrics,
+)
 from ml_peg.app import APP_ROOT
 from ml_peg.app.base_app import BaseApp
 from ml_peg.app.utils.build_callbacks import (
     plot_from_table_column,
     struct_from_scatter,
 )
 from ml_peg.app.utils.load import read_plot
-from ml_peg.models import current_models
-from ml_peg.models.get_models import get_model_names
 
-# Get all models
-MODELS = get_model_names(current_models)
 BENCHMARK_NAME = "Elemental Slab Oxygen Adsorption"
 DOCS_URL = "https://ddmms.github.io/ml-peg/user_guide/benchmarks/surfaces.html#elemental-slab-oxygen-adsorption"
 DATA_PATH = APP_ROOT / "data" / "surfaces" / "elemental_slab_oxygen_adsorption"
@@ -26,25 +29,29 @@
 class ElementalSlabOxygenAdsorptionApp(BaseApp):
     """Elemental slab oxygen adsorption benchmark app layout and callbacks."""
 
-    def register_callbacks(self) -> None:
-        """Register callbacks to app."""
-        scatter = read_plot(
-            DATA_PATH / "figure_adsorption_energies.json",
-            id=f"{BENCHMARK_NAME}-figure",
+    def set_data(self) -> None:
+        """Set data for app."""
+        self.data = read_plot(
+            DATA_PATH / "figure_adsorption_energies.json", id=f"{BENCHMARK_NAME}-figure"
         )
+        self.set_partial_elements()
 
-        structs_dir = DATA_PATH / MODELS[0]
+    def register_callbacks(self) -> None:
+        """Register callbacks to app."""
+        if not hasattr(self, "data") or self.data is None:
+            self.set_data()
 
         # Assets dir will be parent directory
+        structs_dir = DATA_PATH / "mock"
         structs = [
-            f"/assets/surfaces/elemental_slab_oxygen_adsorption/{MODELS[0]}/{struct_file.stem}.xyz"
+            f"/assets/surfaces/elemental_slab_oxygen_adsorption/mock/{struct_file.stem}.xyz"
             for struct_file in sorted(structs_dir.glob("*.xyz"))
         ]
 
         plot_from_table_column(
             table_id=self.table_id,
             plot_id=f"{BENCHMARK_NAME}-figure-placeholder",
-            column_to_plot={"MAE": scatter},
+            column_to_plot={"MAE": self.data},
         )
 
         struct_from_scatter(
@@ -54,6 +61,84 @@ def register_callbacks(self) -> None:
             mode="traj",
         )
 
+    def set_partial_elements(self) -> None:
+        """Get lists of element sets for partial filtering."""
+        try:
+            self.partial_elements = [
+                set(elements) for elements in self.info["elements"]
+            ]
+        except (AttributeError, KeyError, TypeError, IndexError) as err:
+            self.partial_elements = set()
+            warnings.warn(
+                f"Unable to read elements lists for {self.name}: {err}", stacklevel=2
+            )
+
+    def filter_table(
+        self,
+        filter_elements: list[str],
+    ) -> dict[str, dict]:
+        """
+        Filter data by elements.
+
+        Parameters
+        ----------
+        filter_elements
+            List of elements to filter out of data.
+
+        Returns
+        -------
+        dict[str, dict]
+            Updated benchmark table.
+        """
+        # Ensure scatter data and partial elements are set
+        if not hasattr(self, "data") or self.data is None:
+            self.set_data()
+
+        filter_elements = set(filter_elements) if filter_elements else set()
+        table_data = deepcopy(self.table.data)
+
+        # If full overlap, set to None as with basic filtering
+        if not bool(self.elements - filter_elements):
+            for row in table_data:
+                for metric in self.metrics:
+                    row[metric] = None
+            return table_data
+
+        # If no elements filtered, return original table data
+        if not filter_elements:
+            for current_row, original_row in zip(
+                table_data, self.original_table.data, strict=True
+            ):
+                for metric in self.metrics:
+                    current_row[metric] = original_row[metric]
+            return table_data
+
+        # Partial filtering
+        # Get overlap of deselected elements with each system's elements
+        filtered_indices = [
+            not bool(elements & filter_elements) for elements in self.partial_elements
+        ]
+
+        results = {}
+        ref_filtered = False
+
+        for plot in self.data.figure.data:
+            # Ignore unamed (parity) line
+            if plot.name and len(plot.x) != 0:
+                results[plot.name] = np.array(plot.x)[filtered_indices].tolist()
+                if not ref_filtered:
+                    results["ref"] = np.array(plot.y)[filtered_indices].tolist()
+                    ref_filtered = True
+
+        new_metrics = get_metrics(results)
+
+        for row in table_data:
+            model = row["MLIP"]
+            for metric in self.metrics:
+                row[metric] = new_metrics[metric].get(model, None)
+
+        return table_data
+
 
 def get_app() -> ElementalSlabOxygenAdsorptionApp:
     """